使用 Google.Cloud.BigQuery.V2 的 BigQuery 加载作业的幂等性

Question

您可以使用 Google.Cloud.BigQuery.V2 中具有 CreateLoadJob 方法的 BigQueryClient 创建 csv 加载作业，以从 Google Cloud Storage 中的 csv 文件加载数据。

您如何使用此 API 保证幂等性，以确保在收到响应之前说网络中断，并且您开始重试，您最终不会多次将相同的数据加载到 BigQuery 中？

API 使用示例

    private void LoadCsv(string sourceUri, string tableId, string timePartitionField)
    {
        var tableReference = new TableReference()
        {
            DatasetId = _dataSetId,
            ProjectId = _projectId,
            TableId = tableId
        };

        var options = new CreateLoadJobOptions
        {
            WriteDisposition = WriteDisposition.WriteAppend,
            CreateDisposition = CreateDisposition.CreateNever,
            SkipLeadingRows = 1,
            SourceFormat = FileFormat.Csv,
            TimePartitioning = new TimePartitioning
            {
                Type = _partitionByDayType,
                Field = timePartitionField
            }
        };

        BigQueryJob loadJob = _bigQueryClient.CreateLoadJob(sourceUri: sourceUri,
                                                            destination: tableReference,
                                                            schema: null,
                                                            options: options);

        loadJob.PollUntilCompletedAsync().Wait();
        if (loadJob.Status.Errors == null || !loadJob.Status.Errors.Any())
        {
            //Log success
            return;
        }
        //Log error
    }

Answer 1

有两个地方可能最终会丢失响应：

• 创建作业开始时
• 轮询完成时

第一个在没有作业 ID 的情况下相对难以恢复；您可以列出项目中的所有工作，并尝试找到一个与您原本要创建的工作相似的工作。

但是，C# 客户端库会生成一个作业 ID，以便 它 可以重试，或者您可以通过 CreateLoadJobOptions 指定您自己的作业 ID。

第二次失败时间要简单得多：保留返回的BigQueryJob，以便在失败时重试轮询。 （例如，您可以存储作业名称，以便即使您的进程在等待它完成时死亡，您也可以恢复。）

Answer 2

您可以通过基于例如生成自己的 jobid 来实现幂等性。您加载的文件位置和目标表。

job_id = 'my_load_job_{}'.format(hashlib.md5(sourceUri+_projectId+_datasetId+tableId).hexdigest())
var options = new CreateLoadJobOptions
        {
            WriteDisposition = WriteDisposition.WriteAppend,
            CreateDisposition = CreateDisposition.CreateNever,
            SkipLeadingRows = 1,
            JobId = job_id, #add this
            SourceFormat = FileFormat.Csv,
            TimePartitioning = new TimePartitioning
            {
                Type = _partitionByDayType,
                Field = timePartitionField
            }
        };

在这种情况下，如果您尝试重新插入相同的 job_id，则会出错。 如果池化失败，您还可以轻松生成此 job_id 以进行检查。