谷歌云数据流 - 在 bigquery 中批量插入答案

【问题标题】：Google cloud dataflow - batch insert in bigquery谷歌云数据流 - 在 bigquery 中批量插入
【发布时间】：2018-11-23 09:51:26
【问题描述】：

我能够创建一个数据流管道，它从 pub/sub 读取数据，并在处理后以流模式写入大查询。

现在我想以批处理模式运行管道，而不是流模式以降低成本。

目前，我的管道正在使用动态目标在 bigquery 中进行流式插入。我想知道是否有办法使用动态目标执行批量插入操作。

下面是

public class StarterPipeline {  
   public interface StarterPipelineOption extends PipelineOptions {

    /**
     * Set this required option to specify where to read the input.
     */
    @Description("Path of the file to read from")
    @Default.String(Constants.pubsub_event_pipeline_url)
    String getInputFile();

    void setInputFile(String value);

}

@SuppressWarnings("serial")
public static void main(String[] args) throws SocketTimeoutException {

    StarterPipelineOption options = PipelineOptionsFactory.fromArgs(args).withValidation()
            .as(StarterPipelineOption.class);

    Pipeline p = Pipeline.create(options);

    PCollection<String> datastream = p.apply("Read Events From Pubsub",
            PubsubIO.readStrings().fromSubscription(Constants.pubsub_event_pipeline_url));

    PCollection<String> windowed_items = datastream.apply(Window.<String>into(new GlobalWindows())
            .triggering(Repeatedly.forever(
                    AfterProcessingTime.pastFirstElementInPane().plusDelayOf(Duration.standardSeconds(300))))
            .withAllowedLateness(Duration.standardDays(10)).discardingFiredPanes());

    // Write into Big Query
     windowed_items.apply("Read and make event table row", new
     ReadEventJson_bigquery())

     .apply("Write_events_to_BQ",
     BigQueryIO.writeTableRows().to(new DynamicDestinations<TableRow, String>() {
     public String getDestination(ValueInSingleWindow<TableRow> element) {
     String destination = EventSchemaBuilder
     .fetch_destination_based_on_event(element.getValue().get("event").toString());
     return destination;
     }

     @Override
     public TableDestination getTable(String table) {
     String destination =
     EventSchemaBuilder.fetch_table_name_based_on_event(table);
     return new TableDestination(destination, destination);
     }

     @Override
     public TableSchema getSchema(String table) {
     TableSchema table_schema =
     EventSchemaBuilder.fetch_table_schema_based_on_event(table);
     return table_schema;
     }
     }).withCreateDisposition(CreateDisposition.CREATE_NEVER)
     .withWriteDisposition(WriteDisposition.WRITE_APPEND)
     .withFailedInsertRetryPolicy(InsertRetryPolicy.retryTransientErrors()));

    p.run().waitUntilFinish();

    log.info("Events Pipeline Job Stopped");

}

}

【问题讨论】：

标签： google-cloud-platform google-cloud-dataflow

【解决方案1】：

批处理或流式处理由 PCollection 确定，因此您需要将 Pub/Sub 中的数据流 PCollection 转换为批处理 PCollection 以写入 BigQuery。允许这样做的转换是 GroupIntoBatches<K,InputT>。

请注意，由于此转换使用键值对，批次将仅包含单个键的元素。对于非 KV 元素，check this related answer。

使用此转换将 PCollection 创建为批处理后，然后将 BigQuery 写入与动态目标一起应用，就像对流 PCollection 所做的那样。

【讨论】：

使用 Java GroupIntoBatches 不会降低成本，因为仍将执行流式传输。这用于调整批量大小。在批处理管道中，所有记录都在一起。在流式传输管道中，记录在进入管道时或在窗口触发时会经过各个步骤。 GroupIntoBatches 所做的是获取单个记录并按计数对它们进行分组，但管道本身仍在流模式下运行。

【解决方案2】：

您可以使用file loads for Streaming jobs 限制成本。 Insertion Method section 声明 BigQueryIO.Write 支持两种将数据插入 BigQuery 的方法，使用 BigQueryIO.Write.withMethod (org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.Method) 指定。如果未提供任何方法，则将根据输入 PCollection 选择默认方法。有关这些方法的更多信息，请参阅BigQueryIO.Write.Method。

不同的插入方法在成本、配额和数据一致性方面提供了不同的权衡。有关这些权衡的更多信息，请参阅BigQuery documentation。

【讨论】：