WriteToBigQuery 动态表目标返回错误的 tableId

Question

我正在尝试将 bigquery 写入不同的表目标，如果这些表尚不存在，我想动态创建它们。

bigquery_rows | "Writing to Bigquery" >> WriteToBigQuery(lambda e: compute_table_name(e),
                                                schema=compute_table_schema,
                                                additional_bq_parameters=additional_bq_parameters,
                                                write_disposition=BigQueryDisposition.WRITE_APPEND,
                                                create_disposition=BigQueryDisposition.CREATE_IF_NEEDED,
                                                )

compute_table_name 函数其实很简单，我只是想让它工作。

def compute_table_name(element):
    if element['table'] == 'table_id':
        del element['table']
        return "project_id:dataset.table_id"

已正确检测到架构，并已创建表并填充了记录。问题是，我得到的表 ID 类似于：

datasetId: 'dataset'
projectId: 'project_id'
tableId: 'beam_bq_job_LOAD_AUTOMATIC_JOB_NAME_LOAD_STEP...

我也尝试在我的 compute_table_name 函数中返回一个 bigquery.TableReference 对象，但无济于事。

编辑：我正在使用 apache-beam 2.34.0，并且我在 JIRA here

上打开了一个问题

Answer 1

您的管道代码很好。但是，您可以将 callable 传递给 compute_table 名称函数：

bigquery_rows | "Writing to Bigquery" >> WriteToBigQuery(compute_table_name,
                                            schema=compute_table_schema,
                                            additional_bq_parameters=additional_bq_parameters,
                                            write_disposition=BigQueryDisposition.WRITE_APPEND,
                                            create_disposition=BigQueryDisposition.CREATE_IF_NEEDED,
                                            )

BigQuery 中的'beam_bq_job_LOAD_AUTOMATIC_JOB_NAME_LOAD_STEP' 表名可能意味着加载作业尚未完成，或者它有错误；您应该查看 BigQuery 用户界面中的“个人历史记录”或“项目历史记录”标签，了解作业的状态。

Answer 2

我通过关注this answer 找到了我的答案的解决方案。这感觉像是一种解决方法，因为我没有将可调用对象传递给 WriteToBigQuery()。测试了一堆方法，我发现直接给方法提供一个字符串/TableReference 它可以工作，但没有给它一个可调用的。

我每 15 分钟处理约 50 个演出的数据，分布在 6 个表中，并且运行良好。