如何使用“触发一次”触发器控制 Spark 结构化流式处理中每个触发器正在处理的文件量？

Question

我正在尝试使用Spark结构化流的功能，触发一次，来模拟一个类似的批处理设置。然而，当我运行我的初始批处理时，我遇到了一些麻烦，因为我有很多历史数据，因此我也使用了这个选项。选项(" cloud files . includeexistingfiles "，" true ")也处理现有文件。

因此，我的初始批处理变得非常大，因为我无法控制批处理的文件量。

我也尝试过使用选项 cloudFiles.maxBytes触发器，但是，当您使用触发器一次时，这被忽略 -

当我指定最大文件触发器选项时，它也被忽略。它只接受所有可用的文件。

我的代码如下所示:

df = (
  spark.readStream.format("cloudFiles")
    .schema(schemaAsStruct)
    .option("cloudFiles.format", sourceFormat)
    .option("delimiter", delimiter)
    .option("header", sourceFirstRowIsHeader)
    .option("cloudFiles.useNotifications", "true")
    .option("cloudFiles.includeExistingFiles", "true")
    .option("badRecordsPath", badRecordsPath)
    .option("maxFilesPerTrigger", 1)
    .option("cloudFiles.resourceGroup", omitted)
    .option("cloudFiles.region", omitted)
    .option("cloudFiles.connectionString", omitted)
    .option("cloudFiles.subscriptionId", omitted)
    .option("cloudFiles.tenantId", omitted)
    .option("cloudFiles.clientId", omitted)
    .option("cloudFiles.clientSecret", omitted)
    .load(sourceBasePath)
)

# Traceability columns
df = (
  df.withColumn(sourceFilenameColumnName, input_file_name()) 
    .withColumn(processedTimestampColumnName, lit(processedTimestamp))
    .withColumn(batchIdColumnName, lit(batchId))
)

def process_batch(batchDF, id):
  batchDF.persist()
  
  (batchDF
     .write
     .format(destinationFormat)
     .mode("append")
     .save(destinationBasePath + processedTimestampColumnName + "=" +  processedTimestamp)
  )
    
  (batchDF
   .groupBy(sourceFilenameColumnName, processedTimestampColumnName)
   .count()
   .write
   .format(destinationFormat)
   .mode("append")
   .save(batchSourceFilenamesTmpDir))
  
  batchDF.unpersist()

stream = (
  df.writeStream
    .foreachBatch(process_batch)
    .trigger(once=True)
    .option("checkpointLocation", checkpointPath)
    .start()
)

如你所见，我使用的是cloudfiles格式，这是Databricks自动加载器的格式

“Auto Loader在新数据文件到达云存储时增量高效地处理这些文件。

自动加载程序提供了一个名为 cloud 文件的结构化流式处理源。给定云文件存储上的输入目录路径，cloudFiles 源会在新文件到达时自动处理它们，还可以选择处理该目录中的现有文件”

如果我以某种方式以令人困惑的方式提出我的问题，或者它缺乏信息，请说出来。

Answer 1

不幸的是，火花 3.x （DBR

要解决这个问题，您可以定期检查< code > stream . get(' numInputRows ')的值，如果它在一段时间内等于0，则发出< code>stream.stop()

更新，2021 年 10 月：看起来它将在 Spark 3.3 中通过引入新的触发器类型 - Trigger.AvailableNow 进行修复（请参阅 SPARK-36533）