在PySpark SQL中，将具有相同值但不同名称的两列进行合并。

Question

我有两个具有以下结构的表：

Table1：

lang   created_date
java   11-01-23
python 11-11-23

Table2：

lang   ingested_date
scala  11-21-23

我想创建一个合并后的表，期望结果如下：

Table3：

lang   created_date
java   11-01-23
python 11-11-23
scala  11-21-23

但实际得到的结果是：

lang   created_date
java   11-01-23
python 11-11-23
scala  11-21-23
scala  null

我正在使用以下Python PySpark代码，但它在合并两个表时给我多了一行created_date为null的记录。

table1DF = sparkSession.read.table("Table1")
table2DF = sparkSession.read.table("Table2")

table1 = table1DF.select("lang", "created_date")
table2 = table2DF.select("lang", F.col("ingested_date").alias("created_date"))

merged_table = table1.union(table2)
final_table = merged_table.groupBy("lang", "created_date")

如何在使用union合并两个表的数据时避免出现最后一行某个列值为null的情况呢？

Ry- · Answer

您可以使用coalesce函数从两个列中选择非空值。

from pyspark.sql.functions import coalesce

table1DF = sparkSession.read.table("Table1")
table2DF = sparkSession.read.table("Table2")

# 不需要对table2中的列进行重命名，直接合并
table1 = table1DF.select("lang", "created_date")
table2 = table2DF.select("lang", "ingested_date")

merged_table = table1.union(table2)

# 使用coalesce函数处理created_date和ingested_date列，以选取非空值
final_table = merged_table.groupBy("lang").agg(coalesce("created_date", "ingested_date").alias("created_date"))

Billy ONeal · Answer

针对您的问题，我测试了一段代码。union函数并不关心列的名称，它关注的是列的数量和类型，并将数据逐行堆叠在一起。得到一行（'scala', NULL）的情况很奇怪，因为我进行了测试，如果没有在table2中已经存在这一行的话，这个操作应该能正常工作。

from pyspark.sql import SparkSession
from pyspark.sql.types import *
from pyspark.sql import Row
from pyspark.sql.functions import coalesce

spark = SparkSession.builder.appName("SparkSQL").getOrCreate()
# 创建数据样本
schema = StructType([
    StructField("lang", StringType(), True),
    StructField("created_date", StringType(), True)
])
rows = [Row("java", "11-01-23"), Row("python", "11-11-23")]
rows2 = [Row("scala", "11-21-23"), Row("scala", None)]
schema2 = StructType([
    StructField("lang", StringType(), True),
    StructField("ingested_date", StringType(), True)
])

# 创建DataFrame
df = spark.createDataFrame(rows, schema)
df2 = spark.createDataFrame(rows2, schema2)
# 使用union合并DataFrame
result = df.union(df2)
# 删除含有空值的行
result.dropna().show()

即使table1中的第二列名为created_date，而table2中的第二列名为ingested_date，由于union会检查DataFrame的列数量及其每列的数据类型，所以它仍会将数据堆叠起来。

关于您提供的代码，遗憾的是无法复现问题。不过，如果您在执行union后仍然有额外无用的行，您可以这样做：

merged_table = table1.union(table2).dropna()

对此感到好奇并愿意讨论，希望这能帮到您！