Dataflow Pipeline执行报错NotImplementedError，请求排查过滤逻辑

你遇到的RuntimeError: NotImplementedError是因为自定义的FilterStatus1这个DoFn没有遵循Apache Beam的规范——所有自定义的DoFn必须重写process方法，而你写了一个自定义的status_filter_1方法，Beam的运行时无法识别这个方法，所以抛出了未实现的错误。另外还有几个小问题需要一起修正，具体如下：

1. 修正FilterStatus1类的实现 #

Beam的DoFn通过process方法来处理每个输入元素，你需要把过滤逻辑移到这个方法里，同时注意：process方法每次接收的是单个元素（不是集合），所以不需要循环遍历。修正后的代码如下：

class FilterStatus1(beam.DoFn):
    def process(self, element):
        logging.info(f"Processing element: {element}")
        # 直接判断单个元素的status值
        if element["status"] == 1:
            logging.info(f"Keeping element with status=1: {element}")
            yield element

2. 修复Pipeline运行的上下文问题 #

你原来的代码在with beam.Pipeline(...) as p:块结束后才调用p.run()，这会导致错误——因为with块结束后，Pipeline的上下文已经被关闭，无法再启动运行。需要把运行代码移到with块内部：

with beam.Pipeline(options=pipeline_options, argv=pipeline_parameters) as p:
    # Read the pubsub topic into a PCollection.
    lines = (
        p | 'ReadPubSubMessage' >> beam.io.ReadFromPubSub(GOOGLE_PUBSUB_CHANNEL).with_output_types(bytes)
        | 'Decode UTF-8' >> beam.Map(lambda x: x.decode('utf-8'))
        | 'ParsePubSub' >> beam.Map(parse_pubsub)
    )

    (
        lines
        | 'Filter Status 1' >> beam.ParDo(FilterStatus1())
        | 'WriteToBigQueryStatus1' >> beam.io.WriteToBigQuery(
            GOOGLE_BIGQUERY_TABLE
            , project=GOOGLE_PROJECT_ID
            , dataset=GOOGLE_DATASET_ID
            , schema=GoogleBigQuery.get_schema_table(fields_contract)
            , create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED
            , write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND
        )
    )
    logging.info('Pipeline is running...')
    result = p.run()
    result.wait_until_finish()

其他注意点 #

• 确保你的Pub/Sub主题配置正确，能接收到符合格式的JSON消息（比如你给出的{"mac": "KC:FC:48:AE:F6:94", "status": 8, "datetime": "2015-07-13T21:15:02Z"}这类）。
• BigQuery的表结构和fields_contract定义的要匹配，CREATE_IF_NEEDED会自动帮你创建表，但类型要对应正确（比如status是INTEGER类型，确保输入的status值是数字而不是字符串）。

修正完这些后，你的Pipeline应该就能正常过滤status=1的记录，并流式写入BigQuery了。

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