将 Dataflow 与 BigQueryIO.Write.Method.STORAGE_WRITE_API 结合使用的示例

Question

当前的 Dataflow 文档和参考模板（请参阅下面的链接）使用 BigQueryIO.Write.Method.STREAMING_INSERTS 作为 BigQuery 的输入法。

https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/HEAD/src/main/java/com/google/cloud/teleport/templates/PubSubToBigQuery.java

是否有任何代码示例展示了如何将新的STORAGE_WRITE_API 与 Dataflow 一起使用？

Answer 1

您可以从 Apache Beam 存储库中看到一个示例，其中我们有 an integration test。

    Pipeline p = Pipeline.create(options);
    final int payloadSizeBytes = options.getPayloadSizeBytes();

    // Generate input.
    PCollection<Value> values =
        p.apply(
                GenerateSequence.from(1)
                    .to(1000000)
                    .withRate(options.getRecordsPerSecond(), Duration.standardSeconds(1)))
            .apply(
                MapElements.into(TypeDescriptor.of(Value.class))
                    .via(
                        l -> {
                          byte[] payload = "".getBytes(StandardCharsets.UTF_8);
                          if (payloadSizeBytes > 0) {
                            payload = new byte[payloadSizeBytes];
                            ThreadLocalRandom.current().nextBytes(payload);
                          }
                          return new AutoValue_BigQueryStorageAPIStreamingIT_Value(
                              l, ByteBuffer.wrap(payload));
                        }));

    values.apply(
        "writeVortex",
        BigQueryIO.<Value>write()
            .useBeamSchema()
            .to(options.getTargetTable())
            .withMethod(Write.Method.STORAGE_WRITE_API)
            .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
            .withWriteDisposition(WriteDisposition.WRITE_APPEND)
            .withNumStorageWriteApiStreams(options.getNumShards())
            .withTriggeringFrequency(Duration.standardSeconds(options.getTriggerFrequencySec())));

    p.run();