在大查询中写入多个表的数据流中的问题

Question

我使用here的代码

并将其更改为：

def getFullTableName(pn,tn):
    return "{0}:{1}".format(pn,tn)
...
(
  pipeline | "Read Data From Input Topic" >> beam.io.ReadFromPubSub(topic=data_topic)
           | "Get Table data from input row" >> beam.Map(lambda r : data_ingestion.getData(r))
           | "Write to BigQuery Table" >> beam.io.WriteToBigQuery(table = lambda project_name, dest_table_id : getFullTableName(project_name,dest_table_id),                                                                            schema = lambda table, schema_coll : schema_coll[table],                                                                            schema_side_inputs=(schema_coll,),                                                                             create_disposition='CREATE_IF_NEEDED',                                                                           write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND)
        )

运行时出现错误：

来自工作人员的错误消息：generic::unknown: Traceback（最近 最后调用）：文件“apache_beam/runners/common.py”，第 1198 行，在 apache_beam.runners.common.DoFnRunner.process 文件 “apache_beam/runners/common.py”，第 718 行，在 apache_beam.runners.common.PerWindowInvoker.invoke_process 文件 “apache_beam/runners/common.py”，第 843 行，在 apache_beam.runners.common.PerWindowInvoker._invoke_process_per_window 文件 "/usr/local/lib/python3.8/site-packages/apache_beam/io/gcp/bigquery.py", 第 1658 行，正在进行中 schema = self.schema(destination, *schema_side_inputs) 文件“multi_table_stream_dyndest.py”，第 62 行，在 KeyError: 'None:ticketing.test2' 在上述处理期间 异常，发生了另一个异常：Traceback（最近一次调用 最后）：文件 "/usr/local/lib/python3.8/site-packages/apache_beam/runners/worker/sdk_worker.py", 第 267 行，在 _execute response = task() 文件中 "/usr/local/lib/python3.8/site-packages/apache_beam/runners/worker/sdk_worker.py", 第 340 行，在 lambda 中： self.create_worker().do_instruction(request), request) 文件 "/usr/local/lib/python3.8/site-packages/apache_beam/runners/worker/sdk_worker.py", 第 580 行，在 do_instruction return getattr(self, request_type)( File "/usr/local/lib/python3.8/site-packages/apache_beam/runners/worker/sdk_worker.py", 第 618 行，在 process_bundle 中 bundle_processor.process_bundle(instruction_id)) 文件 “/usr/local/lib/python3.8/site-packages/apache_beam/runners/worker/bundle_processor.py”， 第 995 行，在 process_bundle 中 input_op_by_transform_id[element.transform_id].process_encoded( 文件 “/usr/local/lib/python3.8/site-packages/apache_beam/runners/worker/bundle_processor.py”， 第 221 行，在 process_encoded self.output(decoded_value) 文件中 “apache_beam/runners/worker/operations.py”，第 346 行，在 apache_beam.runners.worker.operations.Operation.output 文件 “apache_beam/runners/worker/operations.py”，第 348 行，在 apache_beam.runners.worker.operations.Operation.output 文件 “apache_beam/runners/worker/operations.py”，第 215 行，在 apache_beam.runners.worker.operations.SingletonConsumerSet.receive 文件“apache_beam/runners/worker/operations.py”，第 707 行，在 apache_beam.runners.worker.operations.DoOperation.process 文件 “apache_beam/runners/worker/operations.py”，第 708 行，在 apache_beam.runners.worker.operations.DoOperation.process 文件 “apache_beam/runners/common.py”，第 1200 行，在 apache_beam.runners.common.DoFnRunner.process 文件 “apache_beam/runners/common.py”，第 1281 行，在 apache_beam.runners.common.DoFnRunner._reraise_augmented 文件 “apache_beam/runners/common.py”，第 1198 行，在 apache_beam.runners.common.DoFnRunner.process 文件 “apache_beam/runners/common.py”，第 718 行，在 apache_beam.runners.common.PerWindowInvoker.invoke_process 文件 “apache_beam/runners/common.py”，第 843 行，在 apache_beam.runners.common.PerWindowInvoker._invoke_process_per_window 文件 "/usr/local/lib/python3.8/site-packages/apache_beam/io/gcp/bigquery.py", 第 1658 行，正在进行中 schema = self.schema(destination, *schema_side_inputs) 文件“multi_table_stream_dyndest.py”，第 62 行，在 KeyError 中：“None:ticketing.test2 [while running 'Write to 大查询 表/_StreamToBigQuery/StreamInsertRows/ParDo(BigQueryWriteFn)-ptransform-671']"

有人可以帮我吗？

Answer 1

您的函数正在返回一个错误的表名，您可以在此处看到：

KeyError: "None:ticketing.test2

这意味着，变量project_name 为空（None）。您必须事先设置它，或者从您发送的数据中获取它。与您下面的示例一样，数据具有 tablename 字段：

{"tablename" : "data-analytics-bk:da_belgium_dataset.cust_data", ...

Answer 2

我添加功能： def getFullTableName(self): logging.info('>>> tablename= %s', "{0}:{1}".format(parseInputData.project,parseInputData.tablename)) return "{0}:{1}".format(parseInputData.project,parseInputData.tablename)

并更改 WriteToBigquery： ... table = lambda l : data_ingestion.getFullTableName()

这解决了问题