问如何使用气流将bigquery导出到bigtable？图式问题

EN

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

row_key      | 1_cnt | 2_cnt | 3_cnt
1#2021-08-03 |   1   |   2   |   2 
2#2021-08-02 |   5   |   1   |   5 
.
.
.

"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