从 Dataflow 写入 BigQuery

使用集合让一切井井有条 根据您的偏好保存内容并对其进行分类。

import com.google.api.services.bigquery.model.TableRow; import java.util.Arrays; import java.util.List; import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.coders.DefaultCoder; import org.apache.beam.sdk.extensions.avro.coders.AvroCoder; import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO; import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write; import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition; import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition; import org.apache.beam.sdk.options.PipelineOptionsFactory; import org.apache.beam.sdk.transforms.Create;  public class BigQueryWrite {   // A custom datatype for the source data.   @DefaultCoder(AvroCoder.class)   public static class MyData {     public String name;     public Long age;      public MyData() {}      public MyData(String name, Long age) {       this.name = name;       this.age = age;     }   }    public static void main(String[] args) {     // Example source data.     final List<MyData> data = Arrays.asList(         new MyData("Alice", 40L),         new MyData("Bob", 30L),         new MyData("Charlie", 20L)     );      // Parse the pipeline options passed into the application. Example:     //   --projectId=$PROJECT_ID --datasetName=$DATASET_NAME --tableName=$TABLE_NAME     // For more information, see https://beam.apache.org/documentation/programming-guide/#configuring-pipeline-options     PipelineOptionsFactory.register(ExamplePipelineOptions.class);     ExamplePipelineOptions options = PipelineOptionsFactory.fromArgs(args)         .withValidation()         .as(ExamplePipelineOptions.class);      // Create a pipeline and apply transforms.     Pipeline pipeline = Pipeline.create(options);     pipeline         // Create an in-memory PCollection of MyData objects.         .apply(Create.of(data))         // Write the data to an exiting BigQuery table.         .apply(BigQueryIO.<MyData>write()             .to(String.format("%s:%s.%s",                 options.getProjectId(),                 options.getDatasetName(),                 options.getTableName()))             .withFormatFunction(                 (MyData x) -> new TableRow().set("user_name", x.name).set("age", x.age))             .withCreateDisposition(CreateDisposition.CREATE_NEVER)             .withWriteDisposition(WriteDisposition.WRITE_APPEND)             .withMethod(Write.Method.STORAGE_WRITE_API));     pipeline.run().waitUntilFinish();   } }

import com.google.api.services.bigquery.model.TableFieldSchema; import com.google.api.services.bigquery.model.TableRow; import com.google.api.services.bigquery.model.TableSchema; import java.util.Arrays; import java.util.List; import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.coders.DefaultCoder; import org.apache.beam.sdk.extensions.avro.coders.AvroCoder; import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO; import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write; import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition; import org.apache.beam.sdk.options.PipelineOptionsFactory; import org.apache.beam.sdk.transforms.Create;  public class BigQueryWriteWithSchema {   // A custom datatype for the source data.   @DefaultCoder(AvroCoder.class)   public static class MyData {     public String name;     public Long age;      public MyData() {}      public MyData(String name, Long age) {       this.name = name;       this.age = age;     }   }    public static void main(String[] args) {     // Example source data.     final List<MyData> data = Arrays.asList(         new MyData("Alice", 40L),         new MyData("Bob", 30L),         new MyData("Charlie", 20L)     );      // Define a table schema. A schema is required for write disposition CREATE_IF_NEEDED.     TableSchema schema = new TableSchema()         .setFields(             Arrays.asList(                 new TableFieldSchema()                     .setName("user_name")                     .setType("STRING")                     .setMode("REQUIRED"),                 new TableFieldSchema()                     .setName("age")                     .setType("INT64") // Defaults to NULLABLE             )         );      // Parse the pipeline options passed into the application. Example:     //   --projectId=$PROJECT_ID --datasetName=$DATASET_NAME --tableName=$TABLE_NAME     // For more information, see https://beam.apache.org/documentation/programming-guide/#configuring-pipeline-options     PipelineOptionsFactory.register(ExamplePipelineOptions.class);     ExamplePipelineOptions options = PipelineOptionsFactory.fromArgs(args)         .withValidation()         .as(ExamplePipelineOptions.class);      // Create a pipeline and apply transforms.     Pipeline pipeline = Pipeline.create(options);     pipeline         // Create an in-memory PCollection of MyData objects.         .apply(Create.of(data))         // Write the data to a new or existing BigQuery table.         .apply(BigQueryIO.<MyData>write()             .to(String.format("%s:%s.%s",                 options.getProjectId(),                 options.getDatasetName(),                 options.getTableName()))             .withFormatFunction(                 (MyData x) -> new TableRow().set("user_name", x.name).set("age", x.age))             .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)             .withSchema(schema)             .withMethod(Write.Method.STORAGE_WRITE_API)         );     pipeline.run().waitUntilFinish();   } }

import com.google.api.services.bigquery.model.TableRow; import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.PipelineResult; import org.apache.beam.sdk.coders.StringUtf8Coder; import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO; import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write; import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition; import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition; import org.apache.beam.sdk.options.PipelineOptionsFactory; import org.apache.beam.sdk.testing.TestStream; import org.apache.beam.sdk.transforms.MapElements; import org.apache.beam.sdk.values.TimestampedValue; import org.apache.beam.sdk.values.TypeDescriptor; import org.apache.beam.sdk.values.TypeDescriptors; import org.joda.time.Duration; import org.joda.time.Instant;  public class BigQueryStreamExactlyOnce {   // Create a PTransform that sends simulated streaming data. In a real application, the data   // source would be an external source, such as Pub/Sub.   private static TestStream<String> createEventSource() {     Instant startTime = new Instant(0);     return TestStream.create(StringUtf8Coder.of())         .advanceWatermarkTo(startTime)         .addElements(             TimestampedValue.of("Alice,20", startTime),             TimestampedValue.of("Bob,30",                 startTime.plus(Duration.standardSeconds(1))),             TimestampedValue.of("Charles,40",                 startTime.plus(Duration.standardSeconds(2))),             TimestampedValue.of("Dylan,Invalid value",                 startTime.plus(Duration.standardSeconds(2))))         .advanceWatermarkToInfinity();   }    public static PipelineResult main(String[] args) {     // Parse the pipeline options passed into the application. Example:     //   --projectId=$PROJECT_ID --datasetName=$DATASET_NAME --tableName=$TABLE_NAME     // For more information, see https://beam.apache.org/documentation/programming-guide/#configuring-pipeline-options     PipelineOptionsFactory.register(ExamplePipelineOptions.class);     ExamplePipelineOptions options = PipelineOptionsFactory.fromArgs(args)         .withValidation()         .as(ExamplePipelineOptions.class);     options.setStreaming(true);      // Create a pipeline and apply transforms.     Pipeline pipeline = Pipeline.create(options);     pipeline         // Add a streaming data source.         .apply(createEventSource())         // Map the event data into TableRow objects.         .apply(MapElements             .into(TypeDescriptor.of(TableRow.class))             .via((String x) -> {               String[] columns = x.split(",");               return new TableRow().set("user_name", columns[0]).set("age", columns[1]);             }))         // Write the rows to BigQuery         .apply(BigQueryIO.writeTableRows()             .to(String.format("%s:%s.%s",                 options.getProjectId(),                 options.getDatasetName(),                 options.getTableName()))             .withCreateDisposition(CreateDisposition.CREATE_NEVER)             .withWriteDisposition(WriteDisposition.WRITE_APPEND)             .withMethod(Write.Method.STORAGE_WRITE_API)             // For exactly-once processing, set the triggering frequency.             .withTriggeringFrequency(Duration.standardSeconds(5)))         // Get the collection of write errors.         .getFailedStorageApiInserts()         .apply(MapElements.into(TypeDescriptors.strings())             // Process each error. In production systems, it's useful to write the errors to             // another destination, such as a dead-letter table or queue.             .via(                 x -> {                   System.out.println("Failed insert: " + x.getErrorMessage());                   System.out.println("Row: " + x.getRow());                   return "";                 }));     return pipeline.run();   } }