本教程详细阐述了如何在pyspark中，通过多次条件性左连接（left join）和`coalesce`函数，高效地从一个辅助dataframe中填充主dataframe的缺失值。文章演示了如何针对不同缺失列（如序列号和邮箱）选择不同的连接键，并利用`coalesce`函数优雅地处理优先级和默认值，最终生成一个完整且准确的数据集。
引言
在数据处理和集成任务中，我们经常会遇到需要从一个数据源（辅助DataFrame）中获取信息来补充另一个数据源（主DataFrame）中缺失值的情况。这种补充往往不是简单的合并，而是需要根据主DataFrame中现有数据的缺失情况，选择不同的连接键来查找匹配信息。例如，当主DataFrame的某个标识符缺失时，可能需要通过另一个非标识符字段（如邮箱）去辅助DataFrame中查找该标识符；反之，当邮箱缺失时，可能需要通过已有的标识符去查找邮箱。本教程将详细介绍如何在PySpark环境中，利用多重条件连接和coalesce函数来解决此类复杂的数据填充问题。
数据准备
首先，我们创建两个示例DataFrame：persons作为主DataFrame，其中包含一些缺失的serial_no和mail信息；people作为辅助DataFrame，包含更完整的信息，可用于填充persons中的缺失值。
from pyspark.sql import SparkSessionfrom pyspark.sql.functions import coalesce, lit, col# 初始化SparkSessionspark = SparkSession.builder.appName("FillMissingValues").getOrCreate()# 创建persons DataFramepersons_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_columns = ["name", "age", "serial_no", "mail"]persons = spark.createDataFrame(persons_data, persons_columns)# 创建people DataFramepeople_data = [ ("John", 100483, "john@example.com"), ("Sam", 448900, "sam@example.com"), ("Will", 229809, "will@example.com"), ("Robert", 299011, None), # people 中 Robert 的 mail 也缺失 ("Hill", 567233, "hill@example.com")]people_columns = ["name_p", "s_no", "e_mail"] # 辅助DataFrame列名避免冲突people = spark.createDataFrame(people_data, people_columns)print("原始 persons DataFrame:")persons.show()print("原始 people DataFrame:")people.show()
解决方案：多重连接与Coalesce函数
解决此问题的核心思想是进行多次左连接（Left Join），每次连接都针对一个特定的缺失列，并使用coalesce函数来选择第一个非空值。coalesce函数接受多个列作为参数，并返回第一个非NULL（或None）的值。如果所有参数都为NULL，则返回NULL。在此场景中，我们还会指定一个默认值（”NA”）以防所有匹配失败。
由于需要根据不同的缺失列使用不同的连接键，我们需要分两步进行：
步骤一：填充缺失的serial_no
首先，我们尝试通过mail字段将persons与people连接，以填充persons中缺失的serial_no。
降重鸟
要想效果好，就用降重鸟。AI改写智能降低AIGC率和重复率。
113 查看详情 
# 1. 通过 mail 字段填充 serial_no# 左连接 persons 和 people，连接条件是 persons.mail 等于 people.e_mailserials_enriched = persons.alias("p").join( people.alias("pe"), col("p.mail") == col("pe.e_mail"), "left").select( col("p.name"), col("p.age"), # 使用 coalesce 填充 serial_no： # 优先取 persons 中的 serial_no，如果为 None，则取 people 中的 s_no # 如果两者都为 None，则填充 "NA" coalesce(col("p.serial_no"), col("pe.s_no"), lit("NA")).alias("serial_no"), col("p.mail"))print("填充 serial_no 后的 DataFrame:")serials_enriched.show()
在这一步中，Will的serial_no被成功填充为229809，因为persons.mail与people.e_mail匹配。Robert的serial_no在persons中已经存在，所以保持不变。
步骤二：填充缺失的mail
接下来，我们使用上一步生成的serials_enriched DataFrame，再次与people DataFrame进行左连接，这次使用serial_no字段作为连接键，以填充mail字段的缺失值。
# 2. 通过 serial_no 字段填充 mail# 左连接 serials_enriched 和 people，连接条件是 serials_enriched.serial_no 等于 people.s_nofinal_df = serials_enriched.alias("se").join( people.alias("pe"), col("se.serial_no") == col("pe.s_no"), "left").select( col("se.name"), col("se.age"), col("se.serial_no"), # 使用 coalesce 填充 mail： # 优先取 serials_enriched 中的 mail，如果为 None，则取 people 中的 e_mail # 如果两者都为 None，则填充 "NA" coalesce(col("se.mail"), col("pe.e_mail"), lit("NA")).alias("mail"))print("最终填充后的 DataFrame:")final_df.show()
在这一步中：
Robert的mail在persons中为None。通过serial_no (299011) 与people连接，发现people中Robert对应的e_mail也是None，因此最终填充为NA。Hill的mail在persons中已存在，因此保持不变。
完整代码示例
from pyspark.sql import SparkSessionfrom pyspark.sql.functions import coalesce, lit, col# 初始化SparkSessionspark = SparkSession.builder.appName("FillMissingValues").getOrCreate()# 创建persons DataFramepersons_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_columns = ["name", "age", "serial_no", "mail"]persons = spark.createDataFrame(persons_data, persons_columns)# 创建people DataFramepeople_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_columns = ["name_p", "s_no", "e_mail"]people = spark.createDataFrame(people_data, people_columns)print("--- 原始 DataFrames ---")persons.show()people.show()# --- 步骤一：通过 mail 字段填充 serial_no ---serials_enriched = persons.alias("p").join( people.alias("pe"), col("p.mail") == col("pe.e_mail"), "left").select( col("p.name"), col("p.age"), coalesce(col("p.serial_no"), col("pe.s_no"), lit("NA")).alias("serial_no"), col("p.mail"))print("--- 填充 serial_no 后的 DataFrame ---")serials_enriched.show()# --- 步骤二：通过 serial_no 字段填充 mail ---final_df = serials_enriched.alias("se").join( people.alias("pe"), col("se.serial_no") == col("pe.s_no"), "left").select( col("se.name"), col("se.age"), col("se.serial_no"), coalesce(col("se.mail"), col("pe.e_mail"), lit("NA")).alias("mail"))print("--- 最终填充后的 DataFrame ---")final_df.show()# 停止SparkSessionspark.stop()
注意事项
连接顺序： 本例中，填充serial_no的连接使用了mail字段，而填充mail的连接使用了serial_no字段。如果这两个字段都可能缺失且互为补充，则连接顺序可能影响最终结果。通常，先填充那些能够作为后续连接键的字段会更有效。重复键处理： 如果辅助DataFrame (people) 中存在非唯一连接键（例如，多个e_mail对应同一个s_no，或者多个s_no对应同一个e_mail），左连接可能会导致主DataFrame中的行重复。在实际应用中，如果people DataFrame可能包含重复的e_mail或s_no，建议在连接之前对其进行去重或聚合处理，以确保每个连接键只对应一个补充值。列名冲突： 在进行连接操作时，如果两个DataFrame有同名但含义不同的列，或者连接后希望保留所有列，应使用alias为DataFrame和列指定别名，并在select语句中明确指定要保留的列，避免歧义。coalesce的灵活性： coalesce函数可以接受任意数量的表达式。这意味着您可以定义更复杂的优先级逻辑来填充缺失值，例如从多个辅助数据源中依次尝试获取值。性能考虑： 多次连接操作可能会对性能产生影响，尤其是在处理大规模数据时。优化连接键、选择合适的连接类型以及对DataFrame进行适当的缓存或分区可以帮助提升性能。
总结
通过本教程，我们学习了如何利用PySpark中的多重左连接和coalesce函数来解决复杂的DataFrame缺失值填充问题。这种方法不仅能够根据不同的条件填充不同的列，还能优雅地处理值优先级和默认值设置，是数据清洗和预处理中一项非常实用的技能。理解其背后的逻辑和注意事项，将有助于您在实际项目中更高效、准确地处理数据。
以上就是利用多重条件连接与Coalesce函数填充DataFrame缺失值的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/598978.html
微信扫一扫 
支付宝扫一扫 
