Spark联合大型csv文件并写入镶木地板文件

Question

我有一堆具有相同架构的大型 csv 文件。我想合并这些文件并将结果写入镶木地板文件，按列file_name 分区。

这是我到目前为止所做的：

children_files = hdfs.list("/landing/my_data/children_flow/")
df = spark.createDataFrame(
            spark.sparkContext.emptyRDD(),
            dfSchema
)
df = df.withColumn("file_name", lit(None))
for one_file in children_files :
    df2 = spark.read.csv(os.path.join("/landing/my_data/children_flow/", one_file))
    df2 = df2.withColumn("file_name", lit(one_file.replace(".csv", "")))
    df = df.union(df2)
df.write.partitionBy("file_name").parquet("/staging/my_data/children_flow/")

问题是我收到java.lang.OutOfMemoryError: Java heap space 错误。

我试着做一个

df.repartition(3000, "file_name").write.partitionBy("file_name").parquet("/staging/my_data/children_flow/")

但它也不起作用。您能提出解决方案吗？

Answer 1

HDFS 位置中的文件似乎过多。因为有这么多文件，而union 是一个转换而不是动作，Spark 在尝试构建物理计划时可能会运行 OutOfMemory。

无论如何，如果您打算读取大量具有相同架构的 csv 文件，我会使用结构化流。为避免手动编写架构，您可以从示例文件中推断出它。

下面的代码展示了这个想法：

# define input and output paths and checkpoint location
csvPath = "hdfs:///landing/my_data/children_flow/"
parquetPath = "hdfs:///staging/my_data/children_flow/"
checkpointLoc = "hdfs:///path/to/dir/"

# infer Schema from one example csv file
schemaDf = spark.read.format("csv").option("inferSchema", true).load(csvPath + "first.csv")
schema = schemaDf.schema

# create Stream by reading from csv and writing to parquet
df = spark.readStream \
  .format("csv") \
  .schema(schema) \
  .load(csvPath) \
  .withColumn("file_name", split(input_file_name(), "/")(4))

query = df.writeStream \
  .format("parquet") \
  .outputMode("append") \
  .option("path", parquetPath) \
  .option("checkpointLocation", checkpointLoc) \
  .partitionBy("file_name") \
  .start() \
  .awaitTermination()

如果这再次运行内存不足，您可以将选项“maxFilesPerTrigger”设置为一个合适的数字。如果它创建了太多 parquet 文件，你最终可以只读取 -> 重新分区 -> 写入 parquet 数据一次。