PySpark DataFrame 多条件关联与缺失值填充策略-Python教程-PHP中文网

PySpark DataFrame 多条件关联与缺失值填充策略

本教程详细阐述了在pyspark环境中，如何通过多步条件关联（join）操作，从一个数据框（dataframe）中有效地填充另一个数据框中的缺失值。文章将演示如何根据不同的缺失字段（如序列号或邮箱）选择不同的关联键，并利用`coalesce`函数优雅地处理空值，最终实现数据清洗与整合，确保缺失值被准确填充或标记为“na”。

在数据处理和分析的日常工作中，我们经常会遇到需要从一个数据源补充另一个数据源中缺失信息的情况。当补充的逻辑涉及多个关联键，并且需要根据目标数据框中具体哪个字段缺失来选择不同的关联键时，问题会变得复杂。本教程将以PySpark为例，提供一种清晰、分步的解决方案，通过巧妙地运用多次左连接（Left Join）和coalesce函数来解决此类挑战。

1. 数据准备

首先，我们定义两个示例PySpark DataFrame：persons（主数据框，需要填充缺失值）和 people（参考数据框，提供补充信息）。

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# 初始化SparkSession
spark = SparkSession.builder.appName("FillMissingValues").getOrCreate()

# 定义persons DataFrame
persons_data = [
    ("John", 25, 100483, "john@example.com"),
    ("Sam", 49, 448900, "sam@example.com"),
    ("Will", 63, None, "will@example.com"), # serial_no 缺失
    ("Robert", 20, 299011, None),          # mail 缺失
    ("Hill", 78, None, "hill@example.com")  # serial_no 缺失
]
persons_schema = StructType([
    StructField("name", StringType(), True),
    StructField("age", IntegerType(), True),
    StructField("serial_no", IntegerType(), True),
    StructField("mail", StringType(), True)
])
persons = spark.createDataFrame(persons_data, schema=persons_schema)

# 定义people DataFrame
people_data = [
    ("John", 100483, "john@example.com"),
    ("Sam", 448900, "sam@example.com"),
    ("Will", 229809, "will@example.com"),
    ("Robert", 299011, None),
    ("Hill", 567233, "hill@example.com")
]
people_schema = StructType([
    StructField("name", StringType(), True),
    StructField("s_no", IntegerType(), True),
    StructField("e_mail", StringType(), True)
])
people = spark.createDataFrame(people_data, schema=people_schema)

print("原始 persons DataFrame:")
persons.show()
print("原始 people DataFrame:")
people.show()

登录后复制

原始 persons DataFrame:

+------+---+---------+----------------+
|  name|age|serial_no|            mail|
+------+---+---------+----------------+
|  John| 25|   100483|john@example.com|
|   Sam| 49|   448900| sam@example.com|
|  Will| 63|     NULL|will@example.com|
|Robert| 20|   299011|            NULL|
|  Hill| 78|     NULL|hill@example.com|
+------+---+---------+----------------+

登录后复制

我们的目标是根据people DataFrame中的信息，填充persons DataFrame中serial_no和mail列的缺失值。具体逻辑是：

如果persons中的serial_no缺失，则尝试通过mail列与people进行关联，获取s_no作为补充。
如果persons中的mail缺失，则尝试通过serial_no列与people进行关联，获取e_mail作为补充。
如果关联后仍无匹配值，则填充为“NA”。

2. 核心策略：分步左连接与coalesce函数

由于填充serial_no和mail需要依赖不同的关联键，直接在一个复杂的join语句中实现所有逻辑会非常困难且容易出错。更有效的方法是分两步进行左连接：第一步填充serial_no，第二步在第一步结果的基础上填充mail。

F.coalesce()函数在这里扮演了关键角色，它接受一列或多列作为参数，并返回第一个非NULL的值。这使得我们能够优雅地实现“优先使用现有值，其次使用关联值，最后使用默认值”的逻辑。

2.1 步骤一：填充缺失的序列号 (serial_no)

在此步骤中，我们关注persons DataFrame中serial_no列的缺失值。如果serial_no缺失，我们将尝试通过mail列与people DataFrame的e_mail列进行匹配，以获取s_no作为补充。

喵记多

喵记多 - 自带助理的 AI 笔记

查看详情

# 步骤一：通过 mail 关联填充 serial_no
# 使用左连接，保留 persons 中的所有记录
serials_enriched = persons.join(people, persons.mail == people.e_mail, "left_outer") \
    .select(
        persons.name,
        persons.age,
        F.coalesce(persons.serial_no, people.s_no, F.lit("NA")).alias("serial_no"),
        persons.mail
    )

print("\n填充 serial_no 后的 DataFrame:")
serials_enriched.show()

登录后复制

serials_enriched DataFrame:

+------+---+---------+----------------+
|  name|age|serial_no|            mail|
+------+---+---------+----------------+
|  John| 25|   100483|john@example.com|
|   Sam| 49|   448900| sam@example.com|
|  Will| 63|   229809|will@example.com|
|Robert| 20|   299011|            NULL|
|  Hill| 78|   567233|hill@example.com|
+------+---+---------+----------------+

登录后复制

可以看到，Will和Hill的serial_no已经根据mail成功从people DataFrame中获取并填充。Robert的mail本身就是缺失的，所以这一步无法通过mail关联到s_no，其serial_no保持原样（非缺失）。

2.2 步骤二：填充缺失的邮箱 (mail)

在第一步的基础上，我们现在来处理mail列的缺失值。如果mail缺失，我们将尝试通过serial_no列（现在可能已经包含填充值）与people DataFrame的s_no列进行匹配，以获取e_mail作为补充。

# 步骤二：通过 serial_no 关联填充 mail
# 使用左连接，保留 serials_enriched 中的所有记录
final_df = serials_enriched.join(people, serials_enriched.serial_no == people.s_no, "left_outer") \
    .select(
        serials_enriched.name,
        serials_enriched.age,
        serials_enriched.serial_no,
        F.coalesce(serials_enriched.mail, people.e_mail, F.lit("NA")).alias("mail")
    )

print("\n最终填充后的 DataFrame:")
final_df.show()

登录后复制

最终 final_df DataFrame:

+------+---+---------+----------------+
|  name|age|serial_no|            mail|
+------+---+---------+----------------+
|  John| 25|   100483|john@example.com|
|   Sam| 49|   448900| sam@example.com|
|  Will| 63|   229809|will@example.com|
|Robert| 20|   299011|              NA|
|  Hill| 78|   567233|hill@example.com|
+------+---+---------+----------------+

登录后复制

至此，Will的serial_no和Hill的serial_no都已填充。对于Robert，由于其原始mail缺失，并且在people中Robert对应的e_mail也为NULL，因此最终mail被coalesce函数填充为“NA”，符合预期。

3. 完整示例代码

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# 初始化SparkSession
spark = SparkSession.builder.appName("FillMissingValuesTutorial").getOrCreate()

# 1. 数据准备
persons_data = [
    ("John", 25, 100483, "john@example.com"),
    ("Sam", 49, 448900, "sam@example.com"),
    ("Will", 63, None, "will@example.com"),
    ("Robert", 20, 299011, None),
    ("Hill", 78, None, "hill@example.com")
]
persons_schema = StructType([
    StructField("name", StringType(), True),
    StructField("age", IntegerType(), True),
    StructField("serial_no", IntegerType(), True),
    StructField("mail", StringType(), True)
])
persons = spark.createDataFrame(persons_data, schema=persons_schema)

people_data = [
    ("John", 100483, "john@example.com"),
    ("Sam", 448900, "sam@example.com"),
    ("Will", 229809, "will@example.com"),
    ("Robert", 299011, None),
    ("Hill", 567233, "hill@example.com")
]
people_schema = StructType([
    StructField("name", StringType(), True),
    StructField("s_no", IntegerType(), True),
    StructField("e_mail", StringType(), True)
])
people = spark.createDataFrame(people_data, schema=people_schema)

print("--- 原始数据 ---")
print("persons DataFrame:")
persons.show()
print("people DataFrame:")
people.show()

# 2. 步骤一：通过 mail 关联填充 serial_no
serials_enriched = persons.join(people, persons.mail == people.e_mail, "left_outer") \
    .select(
        persons.name,
        persons.age,
        F.coalesce(persons.serial_no, people.s_no, F.lit("NA")).alias("serial_no"),
        persons.mail
    )

print("--- 步骤一：填充 serial_no 后的 DataFrame ---")
serials_enriched.show()

# 3. 步骤二：通过 serial_no 关联填充 mail
final_df = serials_enriched.join(people, serials_enriched.serial_no == people.s_no, "left_outer") \
    .select(
        serials_enriched.name,
        serials_enriched.age,
        serials_enriched.serial_no,
        F.coalesce(serials_enriched.mail, people.e_mail, F.lit("NA")).alias("mail")
    )

print("--- 最终填充后的 DataFrame ---")
final_df.show()

# 停止SparkSession
spark.stop()

登录后复制

4. 注意事项

数据类型兼容性： 在进行关联操作时，确保用于连接的列（例如 persons.mail 和 people.e_mail）具有兼容的数据类型。coalesce函数也要求其参数列具有兼容的数据类型。
people DataFrame中的重复值： 如果people DataFrame中存在重复的e_mail或s_no值，那么左连接可能会导致persons DataFrame中的记录被重复。例如，如果people中有两条不同的记录都拥有相同的e_mail，那么在第一次连接时，persons中匹配的记录可能会被复制。为避免这种情况，通常建议在进行关联操作前，对people DataFrame进行去重，或者根据业务逻辑选择一个唯一的记录（例如，使用 groupBy().agg(F.first())）。
coalesce的顺序： F.coalesce()函数的参数顺序至关重要。它会按顺序检查每个表达式，并返回第一个非NULL的值。在本例中，F.coalesce(persons.serial_no, people.s_no, F.lit("NA")) 表示：
- 首先尝试使用 persons 中原始的 serial_no。
- 如果 persons.serial_no 为 NULL，则尝试使用 people 中关联到的 s_no。
- 如果两者都为 NULL，则最终使用字符串 "NA"。
性能考量： 对于非常大的数据集，多次连接操作可能会增加计算开销。然而，对于这种需要基于不同键进行条件填充的复杂场景，分步连接通常比尝试构建一个极其复杂的单次连接更清晰、更易维护，且在许多情况下性能表现良好。如果性能成为瓶颈，可以考虑其他高级优化技术，如广播连接（Broadcast Join）或预聚合。

5. 总结

通过本教程，我们学习了如何利用PySpark的DataFrame操作，特别是分步左连接和coalesce函数，来高效且准确地填充数据框中的缺失值。这种方法不仅解决了根据不同缺失字段选择不同关联键的复杂性，而且通过清晰的逻辑和代码结构，提高了数据处理脚本的可读性和可维护性。掌握这种数据整合策略，对于处理实际业务场景中常见的缺失值填充问题至关重要。

以上就是PySpark DataFrame 多条件关联与缺失值填充策略的详细内容，更多请关注php中文网其它相关文章！