PySpark多条件关联与缺失值处理方法

编程并不是一个机械性的工作，而是需要有思考，有创新的工作，语法是固定的，但解决问题的思路则是依靠人的思维，这就需要我们坚持学习和更新自己的知识。今天golang学习网就整理分享《PySpark多条件关联与缺失值处理技巧》，文章讲解的知识点主要包括，如果你对文章方面的知识点感兴趣，就不要错过golang学习网，在这可以对大家的知识积累有所帮助，助力开发能力的提升。

本教程详细阐述了在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作为补充。

# 步骤一：通过 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. 注意事项

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

5. 总结

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

到这里，我们也就讲完了《PySpark多条件关联与缺失值处理方法》的内容了。个人认为，基础知识的学习和巩固，是为了更好的将其运用到项目中，欢迎关注golang学习网公众号，带你了解更多关于的知识点！