【发布时间】:2021-10-10 07:16:00
【问题描述】:
我正在使用 Airflow 以 Avro 格式将 BigQuery 行提取到 Google Cloud Storage。
with models.DAG(
"bigquery_to_bigtable",
default_args=default_args,
schedule_interval=None,
start_date=datetime.now(),
catchup=False,
tags=["test"],
) as dag:
data_to_gcs = BigQueryInsertJobOperator(
task_id="data_to_gcs",
project_id=project_id,
location=location,
configuration={
"extract": {
"destinationUri": gcs_uri, "destinationFormat": "AVRO",
"sourceTable": {
"projectId": project_id, "datasetId": dataset_id,
"tableId": table_id}}})
gcs_to_bt = DataflowTemplatedJobStartOperator(
task_id="gcs_to_bt",
template="gs://dataflow-templates/latest/GCS_Avro_to_Cloud_Bigtable",
location=location,
parameters={
'bigtableProjectId': project_id,
'bigtableInstanceId': bt_instance_id,
'bigtableTableId': bt_table_id,
'inputFilePattern': 'gs://export/test.avro-*'
},
)
data_to_gcs >> gcs_to_bt
bigquery 行包含
row_key | 1_cnt | 2_cnt | 3_cnt
1#2021-08-03 | 1 | 2 | 2
2#2021-08-02 | 5 | 1 | 5
.
.
.
我想将 row_key 列用于 bigtable 中的行键,并将其余列用于特定列族中的列,例如 bigtable 中的 my_cf。
但是我在使用数据流将 avro 文件加载到 bigtable 时收到错误消息
"java.io.IOException: Failed to start reading from source: gs://export/test.avro-"
Caused by: org.apache.avro.AvroTypeException: Found Root, expecting com.google.cloud.teleport.bigtable.BigtableRow, missing required field key
我读到的docs 告诉我们:
Bigtable 表必须存在并且具有相同的列族 在 Avro 文件中导出。
如何在 Avro 中导出具有相同列族的 BigQuery?
【问题讨论】:
标签: google-bigquery dataflow google-cloud-bigtable airflow