Apache Beam Cloud Dataflow Streaming Stuck Side Input

Question

我目前正在 GCP Dataflow 中构建 PoC Apache Beam 管道。在这种情况下，我想使用来自 PubSub 的主要输入和来自 BigQuery 的侧输入来创建流式管道，并将处理后的数据存储回 BigQuery。

侧流水线代码

side_pipeline = (
    p
    | "periodic" >> PeriodicImpulse(fire_interval=3600, apply_windowing=True)
    | "map to read request" >>
        beam.Map(lambda x:beam.io.gcp.bigquery.ReadFromBigQueryRequest(table=side_table))
    | beam.io.ReadAllFromBigQuery()
)

带有侧输入代码的功能

def enrich_payload(payload, equipments):
    id = payload["id"]
    for equipment in equipments:
        if id == equipment["id"]:
            payload["type"] = equipment["type"]
            payload["brand"] = equipment["brand"]
            payload["year"] = equipment["year"]

            break

    return payload

主管道代码

main_pipeline = (
    p
    | "read" >> beam.io.ReadFromPubSub(topic="projects/my-project/topics/topiq")
    | "bytes to dict" >> beam.Map(lambda x: json.loads(x.decode("utf-8")))
    | "transform" >> beam.Map(transform_function)
    | "timestamping" >> beam.Map(lambda src: window.TimestampedValue(
        src,
        dt.datetime.fromisoformat(src["timestamp"]).timestamp()
    ))
    | "windowing" >> beam.WindowInto(window.FixedWindows(30))
)

final_pipeline = (
    main_pipeline
    | "enrich data" >> beam.Map(enrich_payload, equipments=beam.pvalue.AsIter(side_pipeline))
    | "store" >> beam.io.WriteToBigQuery(bq_table)
)

result = p.run()
result.wait_until_finish()

将其部署到 Dataflow 后，一切看起来都很好，没有错误。但后来我注意到enrich data 步骤有两个节点而不是一个。

此外，如您所见，侧面输入卡在输入集合中具有 21 个计数的 Elements Added 和输出集合中的 Elements Added 中的 - 值。

您可以找到完整的管道代码 here 和模拟 pubsub 发布者 here

我已经按照这些文档中的所有说明进行操作：

• https://beam.apache.org/documentation/patterns/side-inputs/
• https://beam.apache.org/releases/pydoc/2.35.0/apache_beam.io.gcp.bigquery.html

但仍然发现此错误。请帮我。谢谢！

Answer 1

这里有一个工作示例：

mytopic = ""
sql = "SELECT station_id, CURRENT_TIMESTAMP() timestamp FROM `bigquery-public-data.austin_bikeshare.bikeshare_stations` LIMIT 10"

def to_bqrequest(e, sql):
    from apache_beam.io import ReadFromBigQueryRequest
    yield ReadFromBigQueryRequest(query=sql)
     

def merge(e, side):
    for i in side:
        yield f"Main {e.decode('utf-8')} Side {i}"

pubsub = p | "Read PubSub topic" >> ReadFromPubSub(topic=mytopic)

side_pcol = (p | PeriodicImpulse(fire_interval=300, apply_windowing=False)
               | "ApplyGlobalWindow" >> WindowInto(window.GlobalWindows(),
                                           trigger=trigger.Repeatedly(trigger.AfterProcessingTime(5)),
                                           accumulation_mode=trigger.AccumulationMode.DISCARDING)
               | "To BQ Request" >> ParDo(to_bqrequest, sql=sql)
               | ReadAllFromBigQuery()
            )

final = (pubsub | "Merge" >> ParDo(merge, side=beam.pvalue.AsList(side_pcol))
                | Map(logging.info)
        )                    
    
p.run()

请注意，这使用GlobalWindow（这样两个输入具有相同的窗口）。我使用了处理时间触发器，以便窗格包含多行。 5 是任意选择的，使用1 也可以。

请注意，侧输入和主输入之间的数据匹配是非确定性的，您可能会看到较旧的触发窗格中的波动值。

理论上，使用FixedWindows 应该可以解决这个问题，但我无法让FixedWindows 工作。