如何调整EventHub入Databricks Delta表流处理批次大小减少小文件

你当前批次写入量被两个参数硬限制，是写入速率低、小文件多的核心原因：

1. setMaxEventsPerTrigger(5) 直接限制每个触发周期最多从EventHub拉取5条事件，这是每次仅写入5条的直接诱因
2. maxBytesPerTrigger=100000（约100KB）的字节限制和事件数限制取最小值生效，进一步压缩了单批次拉取上限

1. 调大EventHub拉取阈值 #

按照你每秒10条的生成速率，10秒触发间隔下单批次至少可以拉取100条，你可以把事件数上限调整到200留冗余，同时对应调大字节上限，避免字节限制先触发。修改后的读流配置示例：

val customEventhubParameters =
  EventHubsConf(connStr.toString())
  .setMaxEventsPerTrigger(200) // 从5调整为200，匹配业务生成速率
val incomingStream = spark.readStream.format("eventhubs").options(customEventhubParameters.toMap).option("maxBytesPerTrigger",1048576).load() // 从100KB调整为1MB

2. 调整触发间隔（可选） #

如果业务对延迟容忍度较高，可以把触发周期从10秒拉长到30秒甚至1分钟，进一步扩大单批次数据量，减少写入次数，从根源减少小文件生成。修改示例：

outputstream.writeStream
  .format("delta")
  .outputMode("append").trigger(Trigger.ProcessingTime("30 seconds")) // 拉长触发间隔
  .option("checkpointLocation", "/temp/checkpoint")
  .start("/temp/eventdata").awaitTermination()

3. 优化Delta写入配置 #

可以在写流参数中开启Delta自带的小文件优化能力，自动合并写入的小文件：

outputstream.writeStream
  .format("delta")
  .outputMode("append").trigger(Trigger.ProcessingTime("10 seconds"))
  .option("checkpointLocation", "/temp/checkpoint")
  .option("optimizeWrite", "true") // 开启写入优化，自动调整写入文件大小
  .option("autoOptimize", "true") // 开启自动优化，后台自动合并小文件
  .start("/temp/eventdata").awaitTermination()

也可以定期手动执行OPTIMIZE delta./temp/eventdata`` 命令合并存量小文件。

4. 减少写入分区数（可选） #

你当前的数据量级很小，单批次数据可以在写之前合并为1个分区，避免多个分区产生多个小文件：

// 在写流前增加重分区逻辑
outputstream.repartition(1).writeStream
  ...// 其余配置不变

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