Scala Spark中如何将两个无关DataFrame按行拼接并输出为单个文件？

问题分析 #

你需要将两个从Hive表生成的无关联DataFrame，按行的顺序一一对应拼接成单行内容，并输出为单个文件。直接用collect()会存在两个问题：一是数据量大时会触发Driver节点内存溢出；二是分布式环境下collect()返回的数组顺序无法保证与原DataFrame的行顺序完全一致，导致拼接结果不符合预期。

解决方案 #

核心思路是给两个DataFrame添加连续自增的行号，通过行号关联实现按行匹配，最后拼接对应内容。具体步骤如下：

1. 导入依赖函数

import org.apache.spark.sql.functions.{row_number, lit, concat_ws}
import org.apache.spark.sql.expressions.Window

2. 为DataFrame添加行号

使用窗口函数row_number()生成连续行号，通过orderBy(lit(1))保证行号按DataFrame默认读取顺序生成（如果原Hive表有可靠排序字段，建议替换lit(1)为该字段，确保顺序绝对稳定）：

// 假设df1是从Hive读取的DataFrame，每行内容对应列名为content1
val df1WithRow = df1.withColumn("row_id", row_number().over(Window.orderBy(lit(1))))

// 同理处理df2，列名为content2
val df2WithRow = df2.withColumn("row_id", row_number().over(Window.orderBy(lit(1))))

3. 关联并拼接内容

通过行号row_id关联两个DataFrame，使用concat_ws()指定分隔符拼接内容：

val joinedDf = df1WithRow.join(df2WithRow, Seq("row_id"), "inner")
  .select(concat_ws("  ", $"content1", $"content2").alias("combined_content"))

• 若两个DataFrame行数不一致，使用"inner"会只保留行数重叠部分；如需保留所有行，可替换为"full_outer"，并通过coalesce()处理空值。

4. 输出为单个文件

使用coalesce(1)将数据合并到一个分区，输出为单个文本文件：

joinedDf.coalesce(1).write.text("/your/output/path")

注意：数据量极大时，coalesce(1)会导致单节点压力过高，可根据实际情况调整分区数，后续手动合并文件。

关键说明 #

• Spark DataFrame本身是无序的，若需严格保证行顺序，必须依赖原表的排序字段（如主键、时间戳）来生成行号，避免使用lit(1)带来的不确定性。
• 避免使用collect()处理大数据集，该操作会将全量数据拉到Driver节点，极易引发内存溢出问题。

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