将数据从 Google Storage 加载到 BigQuery 时如何执行 UPSERT？

Question

BigQuery 支持以下策略：

WRITE_APPEND - 指定可以将行附加到现有表中。

WRITE_EMPTY - 指定输出表必须为空。

WRITE_TRUNCATE - 指定写入应该替换一个表。

它们都不适合UPSERT 操作的目的。

我正在将订单 Json 文件导入 Google 存储，我想将其加载到 BigQuery。正如逻辑所暗示的那样，一些记录将是新的，而其他记录已经存在于以前的加载中并且需要更新（例如更新订单状态（新/暂停/已发送/退款等...）

我正在使用 Airflow，但我的问题很笼统：

update_bigquery = GoogleCloudStorageToBigQueryOperator(
    dag=dag,
    task_id='load_orders_to_BigQuery',
    bucket=GCS_BUCKET_ID,
    destination_project_dataset_table=table_name_template,
    source_format='NEWLINE_DELIMITED_JSON',
    source_objects=[gcs_export_uri_template],
    schema_fields=dc(),
    create_disposition='CREATE_IF_NEEDED',
    write_disposition='WRITE_TRUNCATE',
    skip_leading_rows = 1,
    google_cloud_storage_conn_id=CONNECTION_ID,
    bigquery_conn_id=CONNECTION_ID
)

此代码使用WRITE_TRUNCATE，这意味着它会删除整个表并加载请求的文件。

如何修改它以支持UPSERT？

我唯一的选择是查询表搜索出现在json中的现有订单删除它们然后执行LOAD吗？

Answer 1

MERGE 还不支持DELETE+INSERT *'。如果您想为它加注星标，G' 问题跟踪器中有一个功能请求。

我们也使用 AF 和加载命令 ;-)。因为我们想保留历史变化，所以我们加载到一张表中，然后根据主键字段运行 deDup 查询。 结果保存在单独的表中（截断）。 该表包含我们订单行的最新版本/状态，然后我们将其用于后续查询。

查找 dedup 和 ROW_NUM() 查找 SQL 示例。

请注意，根据数量的不同，您可能不需要具体化，视图或子查询也可以正常工作。

Answer 2

除了运行GoogleCloudStorageToBigQueryOperator之外，您还可以运行一个查询，该查询会为您提供与 upsert 相同的结果。

来自https://cloud.google.com/bigquery/docs/reference/standard-sql/dml-syntax#merge_statement的示例：

MERGE dataset.Inventory T
USING dataset.NewArrivals S
ON T.product = S.product
WHEN MATCHED THEN
  UPDATE SET quantity = T.quantity + S.quantity
WHEN NOT MATCHED THEN
  INSERT (product, quantity) VALUES(product, quantity)

此查询将：

• 查看表 T（当前）和 S（更新）。
• 如果更新更改了现有行，它将在该行上运行 UPDATE。
• 如果更新包含尚不存在的产品，它将INSERT 那个新行。

现在，BigQuery 将如何知道您的表 S？您可以：

• 使用GoogleCloudStorageToBigQueryOperator 将其加载到 BQ 中的另一个表中。
• 或者您可以设置一个直接查看 GCS 的联合表 - 我在 https://medium.com/google-cloud/bigquery-lazy-data-loading-ddl-dml-partitions-and-half-a-trillion-wikipedia-pageviews-cd3eacd657b6 中做过