左连接两个流数据帧然后在 PySpark 中聚合它们后出现空结果

Question

数据说明

我有两个带有日志的CSV文件 - 一个包含跨度日志，另一个包含错误日志（对于熟悉的人来说，弹性APM日志，弹性APM日志）。相关的跨度字段是timestamp，ID，name and code>和持续时间，而相关的错误字段为error_timestamp，error_id和span_id。 span_id错误日志中的字段是一个外键，引用了SPAN日志中的ID字段，从而定义了跨度和错误之间的一对一关系。

注意：我知道，如果我已经有CSV文件，我不应该使用流API，但是我的任务是研究Spark结构化流。连接到实际（生产）流源不是研究阶段的选项，因此我创建了一个Python脚本，该脚本通过编写跨度日志来模拟流源到端口9999，并以相互时间顺序的顺序为端口9998。

任务

我要执行以下

1. ：加入跨度和错误流数据范围使用左外的联接
2. 组由结果范围：
   一个。一个30秒的非重叠时间窗口，通过span.timestamp列窗口
   b。跨度名称
3. 汇总组，计算：
   一个。 跨度的平均值
   b。组中不同的跨度数（span.id）
   c。组中的不同错误数（error.error_id）

我尝试了

模拟流源的服务器当然可以正常工作 - 我尝试处理两个流而无需加入，而一切都很好。

我已经阅读了，而且我认为我主要了解溪流外部连接的原则，水印的目的以及聚合规则。到目前为止，这是我的代码：

from pyspark.sql import functions as F

# ...
# initialization of spans_df and errors_df dataframes
# ...

wm_spans = spans_df.withWatermark("timestamp", "60 seconds")
wm_errors = errors_df.withWatermark("timestamp", "60 seconds")

aggregation = wm_spans.join(
        wm_errors,
        F.expr(
            """
            id = span_id AND
            error_timestamp >= timestamp AND
            error_timestamp <= timestamp + interval 30 seconds
            """
        ), 
        "leftOuter"
    ).groupBy(
        F.window("timestamp", "30 seconds", "30 seconds"), 
        "name"
    ).agg(
        F.avg(F.col('duration')).alias('davg'),
        F.approx_count_distinct(F.col('id')).alias('nSps'),
        F.approx_count_distinct(F.col('error_id')).alias('nErr'),
    )

query = aggregation.writeStream\
    .outputMode("append")\
    .format("console")\
    .start()

预期输出

基于有关窗户聚合中水印的章节和有关外部与水印的章节，不同的跨度名称）对。每个这样的行的错误数可能为0（如果在给定时间窗口中未链接到任何跨度的错误）。

问题

第一次运行时，Spark给了我以下错误：

ERROR UnsupportedOperationChecker: Detected pattern of possible 'correctness' issue due to global 
watermark. The query contains stateful operation which can emit rows older than the current 
watermark plus allowed late record delay, which are "late rows" in downstream stateful operations 
and these rows can be discarded. Please refer the programming guide doc for more details. 
If you understand the possible risk of correctness issue and still need to run the query, 
you can disable this check by setting the config
`spark.sql.streaming.statefulOperator.checkCorrectness.enabled` to false.

因此，我将指定的配置设置为false，这将此错误消息转换为警告消息。但是，结果表是空的。我知道这很可能是由于上述警告中提到的正确性问题，但是我不知道如何解决/避免问题，同时仍达到目标。

我尝试了许多不同的

1. 水印期组合（目前在两个数据范围内60秒）
2. 加入条件期（当前30秒：error_timestamp＆lt; = timestamp＆= timestamp + Interval 30秒）
3. 窗口周期（当前30秒长，长度为30秒， 30秒的间隔）

注意： 并非每个 batch都会给出一个空的结果。起始时间戳为2022-01-31T22：00：00。在22:00:00到22:05:00之间有数千个跨度日志，但只有9个错误日志，其中9个错误在22:04:22和22:04:43之间。因此，它们属于两个30s时窗：22：04：00-22：04：30和22：04：30-22：05：00。处理数据集的这一部分的批次实际上导致了一个看似成功的聚合：

-------------------------------------------------
Batch: 26
-------------------------------------------------
+-----------------------+------------------------+-------...
|start                  |name                    |dmin   ...
+-----------------------+------------------------+-------...
|2022-01-30 22:04:00    |<one of the span names> |3251.0 ...
+-----------------------+------------------------+-------...

我发现这一结果令人惊讶 - 使用左外连接应确保保留所有行。 null列如果在右表中找不到匹配。但是，只有在右表中有匹配项时，我才能得到非空的结果。

任何帮助将不胜感激！如果我忘了陈述一些相关信息，请告诉我，我会尽快添加 - 我不习惯于编写问题，因此请记住这一点。

相关问题

• 数据集涉及聚合 - 这有点相似，但没有帮助，因为OP甚至没有使用水印，这在外部连接两个流数据集时也不是可选的。
• - 主要区别在于，加入两个数据集后，OP未执行分组聚合。相反，他们只是加入他们并使用它们。此外，他们正在使用数据集API，而我正在使用DataFrame API。
• ”同样，此处的OP不使用水印，但他们也试图在非时代的列上使用窗口。