ประมวลผลข้อมูลจำนวนมากด้วย Dataflow

หน้านี้มีตัวอย่างวิธีใช้ Dataflow เพื่อดำเนินการ การดำเนินการ Cloud Firestore จำนวนมากในไปป์ไลน์ Apacheบีม Apache Beam รองรับเครื่องมือเชื่อมต่อสำหรับ Cloud Firestore คุณใช้ เครื่องมือเชื่อมต่อเพื่อเรียกใช้การดำเนินการแบบกลุ่มและสตรีมมิงใน Dataflow

เราขอแนะนำให้ใช้ Dataflow และ Apache Beam สำหรับข้อมูลขนาดใหญ่ การประมวลผลภาระงาน

เครื่องมือเชื่อมต่อ Cloud Firestore สำหรับ Apache Beam จะมีให้ใช้งานใน Java สำหรับข้อมูลเพิ่มเติม ข้อมูลเกี่ยวกับเครื่องมือเชื่อมต่อ Cloud Firestore โปรดดู Apache Beam SDK สำหรับ Java

ก่อนเริ่มต้น

ก่อนที่จะอ่านหน้านี้ คุณควรทำความคุ้นเคยกับ โมเดลการเขียนโปรแกรมสำหรับ Apacheบีม

หากต้องการเรียกใช้ตัวอย่าง คุณต้องเปิดใช้ Dataflow API

ตัวอย่างไปป์ไลน์ Cloud Firestore

ตัวอย่างด้านล่างแสดงไปป์ไลน์ที่เขียนข้อมูลและ อ่านและกรองข้อมูล คุณสามารถใช้ตัวอย่างเหล่านี้เป็นจุดเริ่มต้นสำหรับ ไปป์ไลน์ของตนเอง

การเรียกใช้ไปป์ไลน์ตัวอย่าง

ซอร์สโค้ดสำหรับตัวอย่างมีอยู่ใน googleapis/java-firestore ที่เก็บ GitHub ในการเรียกใช้ตัวอย่างเหล่านี้ ให้ดาวน์โหลดซอร์สโค้ด และดู README

ตัวอย่างไปป์ไลน์ Write

ตัวอย่างต่อไปนี้สร้างเอกสารในคอลเล็กชัน cities-beam-sample



public class ExampleFirestoreBeamWrite {
  private static final FirestoreOptions FIRESTORE_OPTIONS = FirestoreOptions.getDefaultInstance();

  public static void main(String[] args) {
    runWrite(args, "cities-beam-sample");
  }

  public static void runWrite(String[] args, String collectionId) {
    // create pipeline options from the passed in arguments
    PipelineOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(PipelineOptions.class);
    Pipeline pipeline = Pipeline.create(options);

    RpcQosOptions rpcQosOptions =
        RpcQosOptions.newBuilder()
            .withHintMaxNumWorkers(options.as(DataflowPipelineOptions.class).getMaxNumWorkers())
            .build();

    // create some writes
    Write write1 =
        Write.newBuilder()
            .setUpdate(
                Document.newBuilder()
                    // resolves to
                    // projects/<projectId>/databases/<databaseId>/documents/<collectionId>/NYC
                    .setName(createDocumentName(collectionId, "NYC"))
                    .putFields("name", Value.newBuilder().setStringValue("New York City").build())
                    .putFields("state", Value.newBuilder().setStringValue("New York").build())
                    .putFields("country", Value.newBuilder().setStringValue("USA").build()))
            .build();

    Write write2 =
        Write.newBuilder()
            .setUpdate(
                Document.newBuilder()
                    // resolves to
                    // projects/<projectId>/databases/<databaseId>/documents/<collectionId>/TOK
                    .setName(createDocumentName(collectionId, "TOK"))
                    .putFields("name", Value.newBuilder().setStringValue("Tokyo").build())
                    .putFields("country", Value.newBuilder().setStringValue("Japan").build())
                    .putFields("capital", Value.newBuilder().setBooleanValue(true).build()))
            .build();

    // batch write the data
    pipeline
        .apply(Create.of(write1, write2))
        .apply(FirestoreIO.v1().write().batchWrite().withRpcQosOptions(rpcQosOptions).build());

    // run the pipeline
    pipeline.run().waitUntilFinish();
  }

  private static String createDocumentName(String collectionId, String cityDocId) {
    String documentPath =
        String.format(
            "projects/%s/databases/%s/documents",
            FIRESTORE_OPTIONS.getProjectId(), FIRESTORE_OPTIONS.getDatabaseId());

    return documentPath + "/" + collectionId + "/" + cityDocId;
  }
}
ExampleFirestoreBeamWrite.java

ตัวอย่างนี้ใช้อาร์กิวเมนต์ต่อไปนี้ในการกำหนดค่าและเรียกใช้ไปป์ไลน์

GOOGLE_CLOUD_PROJECT=project-id
REGION=region
TEMP_LOCATION=gs://temp-bucket/temp/
NUM_WORKERS=number-workers
MAX_NUM_WORKERS=max-number-workers

ตัวอย่างไปป์ไลน์ Read

ตัวอย่างไปป์ไลน์ต่อไปนี้อ่านเอกสารจาก cities-beam-sample ให้ใช้ตัวกรองสำหรับเอกสารที่มีการตั้งค่าช่อง country เป็น USA และแสดงชื่อของเอกสารที่ตรงกัน



public class ExampleFirestoreBeamRead {

  public static void main(String[] args) {
    runRead(args, "cities-beam-sample");
  }

  public static void runRead(String[] args, String collectionId) {
    FirestoreOptions firestoreOptions = FirestoreOptions.getDefaultInstance();

    PipelineOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(PipelineOptions.class);
    Pipeline pipeline = Pipeline.create(options);

    RpcQosOptions rpcQosOptions =
        RpcQosOptions.newBuilder()
            .withHintMaxNumWorkers(options.as(DataflowPipelineOptions.class).getMaxNumWorkers())
            .build();

    pipeline
        .apply(Create.of(collectionId))
        .apply(
            new FilterDocumentsQuery(
                firestoreOptions.getProjectId(), firestoreOptions.getDatabaseId()))
        .apply(FirestoreIO.v1().read().runQuery().withRpcQosOptions(rpcQosOptions).build())
        .apply(
            ParDo.of(
                // transform each document to its name
                new DoFn<RunQueryResponse, String>() {
                  @ProcessElement
                  public void processElement(ProcessContext c) {
                    c.output(Objects.requireNonNull(c.element()).getDocument().getName());
                  }
                }))
        .apply(
            ParDo.of(
                // print the document name
                new DoFn<String, Void>() {
                  @ProcessElement
                  public void processElement(ProcessContext c) {
                    System.out.println(c.element());
                  }
                }));

    pipeline.run().waitUntilFinish();
  }

  private static final class FilterDocumentsQuery
      extends PTransform<PCollection<String>, PCollection<RunQueryRequest>> {

    private final String projectId;
    private final String databaseId;

    public FilterDocumentsQuery(String projectId, String databaseId) {
      this.projectId = projectId;
      this.databaseId = databaseId;
    }

    @Override
    public PCollection<RunQueryRequest> expand(PCollection<String> input) {
      return input.apply(
          ParDo.of(
              new DoFn<String, RunQueryRequest>() {
                @ProcessElement
                public void processElement(ProcessContext c) {
                  // select from collection "cities-collection-<uuid>"
                  StructuredQuery.CollectionSelector collection =
                      StructuredQuery.CollectionSelector.newBuilder()
                          .setCollectionId(Objects.requireNonNull(c.element()))
                          .build();
                  // filter where country is equal to USA
                  StructuredQuery.Filter countryFilter =
                      StructuredQuery.Filter.newBuilder()
                          .setFieldFilter(
                              StructuredQuery.FieldFilter.newBuilder()
                                  .setField(
                                      StructuredQuery.FieldReference.newBuilder()
                                          .setFieldPath("country")
                                          .build())
                                  .setValue(Value.newBuilder().setStringValue("USA").build())
                                  .setOp(StructuredQuery.FieldFilter.Operator.EQUAL))
                          .buildPartial();

                  RunQueryRequest runQueryRequest =
                      RunQueryRequest.newBuilder()
                          .setParent(DocumentRootName.format(projectId, databaseId))
                          .setStructuredQuery(
                              StructuredQuery.newBuilder()
                                  .addFrom(collection)
                                  .setWhere(countryFilter)
                                  .build())
                          .build();
                  c.output(runQueryRequest);
                }
              }));
    }
  }
}
ExampleFirestoreBeamRead.java

ตัวอย่างนี้ใช้อาร์กิวเมนต์ต่อไปนี้ในการกำหนดค่าและเรียกใช้ไปป์ไลน์

GOOGLE_CLOUD_PROJECT=project-id
REGION=region
TEMP_LOCATION=gs://temp-bucket/temp/
NUM_WORKERS=number-workers
MAX_NUM_WORKERS=max-number-workers

ราคา

การเรียกใช้ภาระงาน Cloud Firestore ใน Dataflow จะทำให้มีค่าใช้จ่าย สำหรับการใช้งาน Cloud Firestore และการใช้งาน Dataflow โฟลว์ข้อมูล ระบบจะเรียกเก็บเงินสำหรับทรัพยากรที่งานของคุณใช้ โปรดดู หน้าการกำหนดราคาโฟลว์ข้อมูล เพื่อดูรายละเอียด สำหรับราคาของ Cloud Firestore โปรดดู หน้าราคา

ขั้นตอนถัดไป