Spark Structured Streaming - 输入速率的峰值减少了批处理持续时间答案

【问题标题】：Spark Structured Streaming - Spike in input rate decreases batch durationSpark Structured Streaming - 输入速率的峰值减少了批处理持续时间
【发布时间】：2019-09-25 01:34:40
【问题描述】：

我遇到了一些乍一看 Spark Streaming 新手似乎违反直觉的事情：

当 Spark Structured Streaming 开始处理更多数据时，它的批处理持续时间会减少

这可能不是最准确的图片，但我看到了更清晰的图案..

我可能需要解释批次持续时间到底是什么 - 我的理解是它代表 Spark 处理流式传输的小批次所需的秒数。

接下来，我需要澄清一下Spark如何触发小批量的处理——是基于批处理中的数据量还是时间间隔...

编辑
代码如下。有相当多的“繁重”操作（joins、dropDuplicates、使用 HOF 过滤、udfs ......）。 Sink 和 Source 都是 Azure Eventhubs

# [CONFIGS]
ehConfig = {
'eventhubs.startingPosition': '{"offset": "@latest", "enqueuedTime": null, 'isInclusive': true,'seqNo': -1}',
'eventhubs.maxEventsPerTrigger': 300,
'eventhubs.connectionString'='XXX'}

ehOutputConfig = {
'eventhubs.connectionString'='YYY' ,   
"checkpointLocation": "azure_blob_storage/ABCABC"
}

spark.conf.set("spark.sql.shuffle.partitions", 3)

# [FUNCS]
@udf(TimestampType())
def udf_current_timestamp():
  return datetime.now()

#-----------#
# STREAMING # 
#-----------#

# [STREAM INPUT]
df_stream_input = spark.readStream.format("eventhubs").options(**_ehConfig).load()

# [ASSEMBLY THE DATAFRAME]
df_joined = (df_stream_input
             .withColumn("InputProcessingStarted", current_timestamp().cast("long"))

             # Decode body
             .withColumn("body_decoded", from_json(col("body").cast("string"), schema=_config))

             # Join customer 
             .join(df_batch, ['CUSTOMER_ID'], 'inner')

             # Filtering
             .filter(expr('body_decoded.status NOT IN (0, 4, 32)'))
             .filter(expr('EXISTS(body_decoded.items, item -> item.ID IN (1, 2, 7))'))

             # Deduplication
             .withWatermark('enqueuedTime', '1 day') 
             .dropDuplicates(['CUSTOMER_ID', 'ItemID']) 

             # Join with lookup table
             .join(broadcast(df_lookup), ['OrderType'], 'left') 

             # UDF
             .withColumn('AssembleTimestamp', udf_current_timestamp())

             # Assemble struct 
             .withColumn('body_struct', struct('OrderType', 'OrderID', 'Price', 'StockPile'))

 # [STREAM OUTPUT]
(df_joined
 .select(to_json('body_struct').alias('body'))
 .writeStream
 .format("eventhubs")
 .options(**_ehOutputConfig)
 .trigger(processingTime='2 seconds')
 .start())

【问题讨论】：

你能描述查询的写入路径吗？触发器、接收器和输出模式。流式查询在做什么？它是无状态的还是有状态的？任何聚合或连接？
嗨@JacekLaskowski，我添加了查询代码。有很多事情要做 - 查询包含 2 个连接、dropDuplicates、解析和组装结构、过滤、udf 等。但最后，查询只是逐行处理 => 输出模式是简单的附加.
补充：我们在两个8GB、2个CPU节点上运行（一个是driver，一个是worker）
您可以在问题中包含streamingQuery.explain 吗？
顺便说一句，您是如何获得指标和图表的？

标签： apache-spark spark-streaming spark-structured-streaming

【解决方案1】：

在 Spark 结构化流中，除非您指定触发选项，否则它会在前一个批次完成处理后立即触发新批次。

在带有 Spark Streaming 的早期版本的 Spark 中，我们可以指定批处理持续时间，比如说 5 秒。在这种情况下，它将每 5 秒触发一次微批处理，并处理最近 5 秒到达的数据。如果是kafka，它会得到尚未提交的数据。

【讨论】：