Airflow BigQuery 运算符 - 将一个表复制到另一个表答案

【问题标题】：Airflow BigQuery Operator - Copying One Table to Another TableAirflow BigQuery 运算符 - 将一个表复制到另一个表
【发布时间】：2021-02-26 09:40:09
【问题描述】：

使用 Airflow 中的 BigQueryOperator，如何将 BigQuery 表（具有所有字符串的架构）复制到另一个 BigQuery 表（具有字符串、整数和浮点数的架构）？请注意，table_1 和 table_2 已在 BigQuery 中使用以下架构创建，并且 table_1 和 table_2 的架构不应更改。

table_1 具有以下架构和数据：

Column_1 (string)  Column_2 (string)  Column_3 (string)  Column_4 (string)  Column_5 (string)
ABC                1                  1                  1.5                1
DEF                2                  2                  2.5                2 
HIJ                3                  3                  3.5                3

table_2 具有以下架构和数据：

Column_1 (string)  Column_2 (integer) Column_3 (integer) Column_4 (float)  Column_5 (integer)
ABC                1                  1                  1.5                1
DEF                2                  2                  2.5                2 
HIJ                3                  3                  3.5                3

为完成此任务，我尝试使用以下 BigQueryOperator，但收到错误消息“查询列 2 的类型为 STRING，无法插入到类型为 INT64 的列 Column_2”中。

BigQuery_Task = BigQueryOperator(
     task_id = "Copy_To_New_Table",
     sql = "INSERT `gcp_project.gcp_dataset.table_2` (Column_1, Column_2, Column_3, Column_4, Column_5) SELECT Column_1, Column_2, Column_3, Column_4, Column_5 FROM `gcp_project.gcp_dataset.table_1`", 
     write_disposition = "WRITE_TRUNCATE",
     location = "US", 
     bigquery_conn_id = "conn_id",
     use_legacy_sql = False, 
     dag = dag)

【问题讨论】：

尝试删除：destination_dataset_table = "gcp_project.gcp_dataset.table_1".
谢谢，这有帮助，但现在收到错误消息“查询列 2 的类型为 STRING，无法插入到类型为 INT64 的列 Column_2”中。

标签： python sql google-cloud-platform google-bigquery airflow

【解决方案1】：

首先，去掉这部分调用：

 destination_dataset_table = "gcp_project.gcp_dataset.table_1

然后编写将运行的 SQL 命令。 BigQuery 通常需要显式转换：

INSERT `gcp_project.gcp_dataset.table_2` (Column_1, Column_2, Column_3, Column_4, Column_5)
    SELECT CAST(Column_1 as int64),
           CAST(Column_2 as int64) as Column_2, 
           CAST(Column_3 as int64) as Column_3, 
           CAST(Column_4 as int64) as Column_4, 
           CAST(Column_5 as int64) as Column_5 
    FROM `gcp_project.gcp_dataset.table_1`"

【讨论】：