PySpark：如何合并数据集并添加基于已有格式的新Date列

嘿，我来帮你搞定这个PySpark数据集合并的问题！根据你的需求，我分两种常见场景给你提供解决方案：

场景1：两个数据集行数一致且顺序完全对应 #

如果你的两个数据集是严格按顺序一一对应的（就像你给出的示例那样），我们可以通过添加自增索引列的方式，将两个数据集关联起来，具体步骤如下：

步骤1：创建示例数据集（模拟你的数据） #

from pyspark.sql import SparkSession
from pyspark.sql.functions import monotonically_increasing_id, col

# 初始化Spark会话
spark = SparkSession.builder.appName("MergeDateDatasets").getOrCreate()

# 第一个包含Y、M、D列的数据集
data1 = [(2017, 3, 4), (2017, 2, 5)]
df1 = spark.createDataFrame(data1, ["Y", "M", "D"])

# 第二个包含Y:M:D单列的数据集
data2 = [("2017:3:4",), ("2017:2:5",)]
df2 = spark.createDataFrame(data2, ["Y:M:D"])

步骤2：添加自增索引列 #

给两个数据集都加上唯一的自增索引，用来确保数据可以正确匹配：

df1_with_idx = df1.withColumn("idx", monotonically_increasing_id())
df2_with_idx = df2.withColumn("idx", monotonically_increasing_id())

步骤3：关联数据集并整理结果 #

通过索引列关联两个数据集，然后重命名列并去掉临时的索引列：

# 按索引内连接，选择需要的列并将Y:M:D重命名为Date
result_df = df1_with_idx.join(df2_with_idx, on="idx", how="inner") \
    .select("Y", "M", "D", col("Y:M:D").alias("Date")) \
    .drop("idx")

# 查看最终结果
result_df.show()

执行后你会得到完全符合要求的输出：

+----+---+---+----------+
|   Y|  M|  D|      Date|
+----+---+---+----------+
|2017|  3|  4|2017:3:4|
|2017|  2|  5|2017:2:5|
+----+---+---+----------+

场景2：无需依赖第二个数据集（更高效的方式） #

其实观察你的数据可以发现，第二个数据集的Y:M:D列就是第一个数据集Y、M、D列的拼接结果。如果你的业务场景允许，完全可以直接在第一个数据集上生成Date列，不需要合并第二个数据集，这样效率更高：

from pyspark.sql.functions import concat_ws

# 使用concat_ws函数拼接Y、M、D列，分隔符为冒号
result_df = df1.withColumn("Date", concat_ws(":", col("Y"), col("M"), col("D")))

result_df.show()

这个方法同样能得到你想要的目标数据集，而且省去了数据集关联的步骤，性能更优。

内容的提问来源于stack exchange，提问作者BryceSoker