我正在运行一个 apache beam 数据流作业，它从存储桶中读取数据，执行一些转换并写入 bigquery。 但是记录被插入到流式缓冲区中。

validated_data = (p1
                  | 'Read files from Storage '+url >> beam.io.ReadFromText(url)
                  | 'Validate records ' + url >> beam.Map(data_ingestion.validate, url)\
                  .with_outputs(SUCCESS_TAG_KEY, FAILED_TAG_KEY, main="main")
)
all_data, _, _ = validated_data
success_records = validated_data[SUCCESS_TAG_KEY]
failed_records = validated_data[FAILED_TAG_KEY]


(success_records
 | 'Extracting row from tagged row {}'.format(url) >> beam.Map(lambda row: row['row'])
 | 'Write to BigQuery table for {}'.format(url) >> beam.io.WriteToBigQuery(
            table=data_ingestion.get_table(tmp=TEST, run_date=data_ingestion.run_date),
            create_disposition=beam.io.BigQueryDisposition.CREATE_NEVER,
            write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND
            )
)

实际上，我需要在上面运行之前删除分区，以避免摄取时间分区表的重复记录。

如果我对同一个文件运行此作业超过 1 次，而不截断表格，表格最终将有重复的记录。

并且由于最后的记录在流缓冲区中，删除分区表命令实际上并没有删除分区。 下面是我用来截断表的代码。并且此代码在运行管道之前运行

client = bigquery.Client()
dataset = TABLE_MAP['dataset']
table = TABLE_MAP[sentiment_pipeline][table_type]['table']
table_id = "{}${}".format(table, format_date(run_date, '%Y%m%d'))
table_ref = client.dataset(dataset).table(table_id)
output = client.delete_table(table_ref)

最佳答案

根据 BigQuery 文档，您可能需要等待 30 minutes in order to make a DML statement on a a streaming table ，并且架构更改为 delete/truncate tables might result in data loss for some scenarios . Here您可以尝试一些解决方法来处理流媒体场景中的重复项。

此外，Apache Beam和 Dataflow现在支持 python 的批量插入，因此这可能是避免流限制的好方法。

https://stackoverflow.com/questions/54153333/