python - 从 Python 中的 Apache Beam Dataflow 连接到 Google Cloud BigQuery 时出现 TypeError?
问题描述
当尝试在 apache 梁的谷歌云数据流中初始化 python BigQuery Client() 时,它给了我一个类型错误:
TypeError('__init__() takes 2 positional arguments but 3 were given')
我正在使用 Python 3.7 和 apache Beam 数据流,我必须手动初始化客户端并写入 BigQuery,而不是使用 ptransform,因为我想使用通过运行时参数传递的动态表名。
我尝试将项目和凭据传递给客户端,但它似乎没有做任何事情。此外,如果我使用 google-cloud-bigquery==1.11.2 而不是 1.13.0 它可以正常工作,在 apache Beam 之外使用 1.13.0 也可以完全正常工作。
我显然已经删掉了一些代码,但这基本上是引发错误的原因
class SaveObjectsBigQuery(beam.DoFn):
def process(self, element, *args, **kwargs):
# Establish BigQuery client
client = bigquery.Client(project=project)
def run():
pipeline_options = PipelineOptions()
# GoogleCloud options object
cloud_options = pipeline_options.view_as(GoogleCloudOptions)
pipeline_options.view_as(SetupOptions).save_main_session = True
with beam.Pipeline(options=pipeline_options) as p:
_data = (p
| "Create" >> beam.Create(["Start"])
)
save_data_bigquery = _data | "Save to BigQuery" >> beam.ParDo(SaveObjectsBigQuery())
在早期版本的 google-cloud-bigquery 中,这可以正常工作,我可以毫无问题地使用运行时参数和 insert_rows_json 创建一个表。显然,使用 WriteToBigquery Ptransform 是理想的,但由于动态命名 bigquery 表的必要性,这是不可能的。
编辑:
我更新了代码以尝试取出运行时值提供程序和 lambda 函数,尽管两者都收到了类似的错误:
`AttributeError: 'function/RuntimeValueProvider' 对象没有属性 'tableId'
我本质上是在启动数据流模板时尝试使用运行时值提供程序来使用 WriteToBigQuery PTransform 动态命名 bigquery 表。
save_data_bigquery = _data | WriteToBigQuery(
project=project,
dataset="campaign_contact",
table=value_provider.RuntimeValueProvider(option_name="table", default_value=None, value_type=str),
schema="id:STRING",
create_disposition=BigQueryDisposition.CREATE_IF_NEEDED,
write_disposition=BigQueryDisposition.WRITE_APPEND
)
save_data_bigquery = _data | WriteToBigQuery(
table=lambda table: f"{project}:dataset.{runtime_options.table}",
schema="id:STRING",
create_disposition=BigQueryDisposition.CREATE_IF_NEEDED,
write_disposition=BigQueryDisposition.WRITE_APPEND
)
解决方案
从 Beam 2.12 开始,您可以使用WriteToBigQuery
转换动态分配目标。我建议你试试看:)
在 Beam 代码库中查看此测试,其中显示了此示例。
推荐阅读
- cakephp - 如何从 cakecoded 自定义 ckeditor?
- arrays - 基于一个标准返回多个值的问题
- azure-devops - Azure DevOps 中特定用户的字段级别权限
- javascript - 放大/缩小时传单标记移动
- javascript - 这个错误是什么意思:“resolver::resolvePath ...”?
- javascript - Gatsby CLI 无法安装
- c# - 实体框架无法在实体构造函数中绑定值对象
- ios - 在 v5 中将预取图像复制到内存 (SDWebImageCacheMemoryOnly)
- ide - 如何在 PhpStorm 中的数组周围保留空白行
- python - 在多张 Excel 中加载数据框的问题