如何使用 Apache Beam 从 Google Pub/Sub 访问消息 ID？

我一直在 Python 2.7.16 上使用 2.13.0 SDK 测试 Apache Beam，以流模式从 Google Pub/Sub 订阅中提取简单消息，并写入 Google Big Query 表。作为此操作的一部分，我尝试使用 Pub/Sub 消息 ID 进行重复数据删除，但我似乎根本无法将其取出。

The ReadFromPubSub 方法的文档 https://beam.apache.org/releases/pydoc/2.13.0/apache_beam.io.gcp.pubsub.html#apache_beam.io.gcp.pubsub.ReadFromPubSub and 发布订阅消息类型 https://beam.apache.org/releases/pydoc/2.13.0/apache_beam.io.gcp.pubsub.html#apache_beam.io.gcp.pubsub.PubsubMessage建议服务生成的 KV（例如 id_label）应作为 attribute 属性的一部分返回，但它们似乎并未返回。

请注意，仅在使用数据流运行器时才支持 id_label 参数。

发送消息的代码

import time
import json
from datetime import datetime

from google.cloud import pubsub_v1

project_id = "[YOUR PROJECT]"
topic_name = "test-apache-beam"

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(project_id, topic_name)

def callback(message_future):
    if message_future.exception(timeout=30):
        print ('Publishing message {} threw an Exception {}.'.format(topic_name, message_future.exception()))
    else:
        print(message_future.result())

for n in range(1,11):
    data = {'rownumber':n}
    jsondata = json.dumps(data)
    message_future = publisher.publish(topic_path, data=jsondata, source='python', timestamp=datetime.now().strftime("%Y-%b-%d (%H:%M:%S:%f)"))
    message_future.add_done_callback(callback)

print('Published message IDs:')

Beam 管道代码：-

from __future__ import absolute_import

import argparse
import logging
import re
import json
import time
import datetime
import base64
import pprint

from past.builtins import unicode

import apache_beam as beam
from apache_beam.io import ReadFromText
from apache_beam.io import ReadFromPubSub
from apache_beam.io import ReadStringsFromPubSub
from apache_beam.io import WriteToText
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.pipeline_options import SetupOptions
from apache_beam.options.pipeline_options import StandardOptions
from apache_beam.transforms.trigger import AfterProcessingTime
from apache_beam.transforms.trigger import AccumulationMode

def format_message_element(message, timestamp=beam.DoFn.TimestampParam):

    data = json.loads(message.data)
    attribs = message.attributes

    fullmessage = {'data' : data,
                   'attributes' : attribs,
                   'attribstring' : str(message.attributes)}

    return fullmessage

def run(argv=None):

    parser = argparse.ArgumentParser()
    input_group = parser.add_mutually_exclusive_group(required=True)
    input_group.add_argument(
                        '--input_subscription',
                        dest='input_subscription',
                        help=('Input PubSub subscription of the form '
                        '"projects/<PROJECT>/subscriptions/<SUBSCRIPTION>."'))
    input_group.add_argument(
                        '--test_input',
                        action="store_true",
                        default=False
    )
    group = parser.add_mutually_exclusive_group(required=True) 
    group.add_argument(
      '--output_table',
      dest='output_table',
      help=
      ('Output BigQuery table for results specified as: PROJECT:DATASET.TABLE '
       'or DATASET.TABLE.'))
    group.add_argument(
        '--output_file',
        dest='output_file',
        help='Output file to write results to.')
    known_args, pipeline_args = parser.parse_known_args(argv)

    options = PipelineOptions(pipeline_args)
    options.view_as(SetupOptions).save_main_session = True

    if known_args.input_subscription:
        options.view_as(StandardOptions).streaming=True

    with beam.Pipeline(options=options) as p:

        from apache_beam.io.gcp.internal.clients import bigquery

        table_schema = bigquery.TableSchema()

        attribfield = bigquery.TableFieldSchema()
        attribfield.name = 'attributes'
        attribfield.type = 'record'
        attribfield.mode = 'nullable'

        attribsource = bigquery.TableFieldSchema()
        attribsource.name = 'source'
        attribsource.type = 'string'
        attribsource.mode = 'nullable'

        attribtimestamp = bigquery.TableFieldSchema()
        attribtimestamp.name = 'timestamp'
        attribtimestamp.type = 'string'
        attribtimestamp.mode = 'nullable'

        attribfield.fields.append(attribsource)
        attribfield.fields.append(attribtimestamp)
        table_schema.fields.append(attribfield)

        datafield = bigquery.TableFieldSchema()
        datafield.name = 'data'
        datafield.type = 'record'
        datafield.mode = 'nullable'

        datanumberfield = bigquery.TableFieldSchema()
        datanumberfield.name = 'rownumber'
        datanumberfield.type = 'integer'
        datanumberfield.mode = 'nullable'
        datafield.fields.append(datanumberfield)
        table_schema.fields.append(datafield)

        attribstringfield = bigquery.TableFieldSchema()
        attribstringfield.name = 'attribstring'
        attribstringfield.type = 'string'
        attribstringfield.mode = 'nullable'
        table_schema.fields.append(attribstringfield)

        if known_args.input_subscription:
            messages = (p
            | 'Read From Pub Sub' >> ReadFromPubSub(subscription=known_args.input_subscription,with_attributes=True,id_label='message_id')
            | 'Format Message' >> beam.Map(format_message_element)
            )

            output = (messages | 'write' >> beam.io.WriteToBigQuery(
                        known_args.output_table,
                        schema=table_schema,
                        create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
                        write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND)
                    )

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

if __name__ == '__main__':
  logging.getLogger().setLevel(logging.INFO)
  run()

以及运行 python 脚本的代码：-

python PythonTestMessageId.py --runner DataflowRunner --project [YOURPROJECT] --input_subscription projects/[YOURPROJECT]/subscriptions/test-apache-beam.subscription --output_table [YOURPROJECT]:test.newtest --temp_location gs://[YOURPROJECT]/tmp --job_name test-job

在提供的代码中，我只是将 Attributes 属性的字典转换为字符串，然后插入到 BigQuery 表中。表中返回的数据如下所示：-

正如您所看到的，attributes 字段中的两个属性只是我传入的属性，并且 PubSub 消息 id 不可用。

有没有办法可以退回这个？

这是一个已知的问题。 A错误报告 https://issues.apache.org/jira/browse/BEAM-3489已在 JIRA 中归档，用于在 PubsubMessage 中公开 message_id。请投票支持此错误报告。