气流GCSToBigQueryOperator正在重新排序我的列

Question

我的DAG中有以下操作员。他们从我的MySQL数据库接收数据，将其上传到GCS，然后导入到BigQuery。它运行得很好！有一个小问题..。

我可以看到，在create和import任务之间，目标表是使用schema参数中指定的模式以正确的列顺序在BigQuery中创建的。但是，一旦import任务运行，表的模式就会发生变化，并且将列重新排序为看似任意的排序。为什么会发生这种情况，是否有办法让BigQuery停止这样做？我看到操作符上有schema_update_options，但是文档很差.

            create=BigQueryCreateEmptyTableOperator(
                task_id="create",
                bigquery_conn_id='google_cloud',
                project_id="<myproject>",
                dataset_id=target_dataset,
                table_id=table_name,
                schema_fields=schema
            )
            upload=MySQLToGCSOperator(
                task_id='mysql_to_gcs', 
                mysql_conn_id='bi_mysql',
                sql=self.sql,
                bucket=self.bucket,
                filename=self.filename,
                export_format='NEWLINE_DELIMITED_JSON',
                google_cloud_storage_conn_id='google_cloud'
            )
            import=GCSToBigQueryOperator(
                task_id='gcs_to_bigquery',
                bucket=self.bucket,
                source_format='NEWLINE_DELIMITED_JSON',
                source_objects=[self.filename],
                destination_project_dataset_table="<myproject>..target_dataset.{table_name}",
                write_disposition='WRITE_TRUNCATE',
                bigquery_conn_id='google_cloud',
                google_cloud_storage_conn_id='google_cloud',
            )
            
            create >> upload >> import

Answer 1

重新排序的发生是因为您没有在您的schema_fields中定义GCSToBigQueryOperator，并且它触发了BigQuery模式自动检测，其中

BigQuery尽最大努力从源数据中自动推断架构。

在您的情况下，为了确保按您希望的方式定义列的顺序，您必须在您的schema_fields中定义GCSToBigQueryOperator。

您可以省略BigQueryCreateEmptyTableOperator，因为GCSToBigQueryOperator已经可以创建BigQuery表并定义模式。

请参阅根据您发布的问题更新的代码：

upload=MySQLToGCSOperator(
                task_id='mysql_to_gcs', 
                mysql_conn_id='bi_mysql',
                sql=self.sql,
                bucket=self.bucket,
                filename=self.filename,
                export_format='NEWLINE_DELIMITED_JSON',
                google_cloud_storage_conn_id='google_cloud'
            )
            create_and_import=GCSToBigQueryOperator(
                task_id='gcs_to_bigquery',
                bucket=self.bucket,
                source_format='NEWLINE_DELIMITED_JSON',
                source_objects=[self.filename],
                destination_project_dataset_table="<myproject>..target_dataset.{table_name}",
                write_disposition='WRITE_TRUNCATE',
                bigquery_conn_id='google_cloud',
                google_cloud_storage_conn_id='google_cloud',
                schema_fields=schema
            )
            
            upload >> create_and_import

有关更多细节，您可以参考此GCSToBigQueryOperator文档。