如何用结构化流的writestream进行分区的文件写入？

Question

我有一个结构化的流代码，可以从Kafka读取数据并将其转储到HDFS。转储数据时，我根据三列对数据进行分区。我面临的问题是在批处理过程中生成了许多小文件。我想在每个partitionBy的批处理期间仅生成一个文件。我不确定如何在这种情况下应用分区，因为它似乎不起作用。

        query = df.selectExpr("CAST(value as STRING)") \
                .repartition(1) \
                .writeStream.partitionBy('host', 'dt', 'h') \ ==> repartition(1) is not working here
                .format("parquet") \
                .outputMode("append") \
                .option("checkpointLocation", self.checkpoint_location) \
                .option('path', self.hdfs_path) \
                .option('failOnDataLoss', 'false') \
                .option("startingOffset", "earliest") \
                .trigger(processingTime='2 seconds').start()

我不想编写另一个清理作业，该作业从路径中读取数据，对其进行重新分区，并在每个分区中存储具有所需文件数的数据。

Answer 1

我已经使用分区进行了一些测试，它似乎对我有用。我创建了一个测试Kafka主题，它具有字符串格式id-value的数据。然后，在流式代码中，我在value上拆分-，并用partitionBy('id')写入数据以模仿您的代码行为。我正在使用kafka经纪人0.10和Spark版本2.4.3。

请参见下面的代码：

from pyspark.sql.functions import col, split
df = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "localhost:9092") \
  .option("subscribe", "partitionTestTopic") \
  .option("startingOffsets", "earliest") \
  .load()

使用repartition(1)：

df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)") \
  .withColumn('id', split(col('value'), '-').getItem(0)) \
  .repartition(1) \
  .writeStream.partitionBy('id') \
  .format("parquet") \
  .outputMode("append") \
  .option('path', 'test-1/data') \
  .option("checkpointLocation", "test-1/checkpoint") \
  .trigger(processingTime='20 seconds') \
  .start()

输出：

├── id=1
│   └── part-00000-9812bd07-3c0f-442e-a80c-5c09553f20e8.c000.snappy.parquet
├── id=2
│   └── part-00000-522e99b6-3900-4702-baf7-2c55819b775c.c000.snappy.parquet
├── id=3
│   └── part-00000-5ed9bef0-4941-451f-884e-8e94a351323f.c000.snappy.parquet

使用repartition(3)：

df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)") \
  .withColumn('id', split(col('value'), '-').getItem(0)) \
  .repartition(3) \
  .writeStream.partitionBy('id') \
  .format("parquet") \
  .outputMode("append") \
  .option('path', 'test-3/data') \
  .option("checkpointLocation", "test-3/checkpoint") \
  .trigger(processingTime='20 seconds') \
  .start()

输出：

├── id=1
│   ├── part-00000-ed6ed5dd-b376-40a2-9920-d0cb36c7120f.c000.snappy.parquet
│   ├── part-00001-fa64e597-a4e1-4ac2-967f-5ea8aae96c13.c000.snappy.parquet
│   └── part-00002-0e0feab8-57d8-4bd2-a94f-0206ff90f16e.c000.snappy.parquet
├── id=2
│   ├── part-00000-c417dac5-271f-4356-b577-ff6c9f45792e.c000.snappy.parquet
│   ├── part-00001-7c90eb8a-986a-4602-a386-50f8d6d85e77.c000.snappy.parquet
│   └── part-00002-0e59e779-84e8-4fcf-ad62-ef3f4dbaccd5.c000.snappy.parquet
├── id=3
│   ├── part-00000-8a555649-1141-42fe-9cb5-0acf0efc5997.c000.snappy.parquet
│   ├── part-00001-ce4aaa50-e41b-4f5f-837c-661459b747b8.c000.snappy.parquet
│   └── part-00002-9f95261e-bd4c-4f1e-bce2-f8ab3b8b01ec.c000.snappy.parquet

您提到您正在使用10-minute批次，因此也应相应地更新触发间隔。现在是2 seconds，但应该是trigger(processingTime='10 minutes')（请参阅class ProcessingTime(Trigger) here）。这可能就是为什么您得到太多小文件的原因。

如果将reaprtition(1)与10 minute batch一起使用，将会有大量的数据混排，并且只有一个内核（每个host, dt and h）最终将写入所有数据，并且您将无法使用Spark并行处理来范围。还有一个缺点是比128MB大的分区。由于您没有在查询中进行任何聚合，因此应该decrease your batch size并使用repartition(1)，以便可以具有可接受的分区大小。