Xử lý dữ liệu hàng loạt bằng Dataflow

Trang này đưa ra ví dụ về cách sử dụng Dataflow để thực hiện các hoạt động hàng loạt trên Cloud Firestore trong một quy trình Apache Beam. Apache Beam hỗ trợ trình kết nối dành cho Cloud Firestore. Bạn có thể sử dụng trình kết nối này để chạy các thao tác hàng loạt và truyền trực tuyến trong Dataflow.

Bạn nên sử dụng Dataflow và Apache Beam cho các tải công việc xử lý dữ liệu có quy mô lớn.

Trình kết nối Cloud Firestore dành cho Apache Beam có trong Java. Để biết thêm thông tin về trình kết nối Cloud Firestore, hãy xem SDK Apache báo cho Java.

Trước khi bắt đầu

Trước khi đọc trang này, bạn nên làm quen với Mô hình lập trình cho Apache Beam.

Để chạy các mẫu, bạn phải bật Dataflow API.

Ví dụ về quy trình Cloud Firestore

Các ví dụ bên dưới minh hoạ một quy trình ghi dữ liệu, còn một quy trình đọc và lọc dữ liệu. Bạn có thể sử dụng các mẫu này làm điểm xuất phát cho các quy trình của riêng mình.

Chạy quy trình mẫu

Mã nguồn cho các mẫu có sẵn trong googleapis/java-firestore kho lưu trữ GitHub. Để chạy các mẫu này, hãy tải mã nguồn xuống và xem tệp README.

Ví dụ về quy trình Write

Ví dụ sau đây sẽ tạo tài liệu trong tập hợp 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

Ví dụ này sử dụng các đối số sau để định cấu hình và chạy một quy trình:

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

Ví dụ về quy trình Read

Quy trình mẫu sau sẽ đọc tài liệu từ bộ sưu tập cities-beam-sample, áp dụng bộ lọc cho các tài liệu trong đó trường country được đặt thành USA và trả về tên của các tài liệu trùng khớp.



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

Ví dụ này sử dụng các đối số sau để định cấu hình và chạy một quy trình:

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

Giá

Việc chạy khối lượng công việc trên Cloud Firestore trong Dataflow sẽ làm phát sinh chi phí sử dụng Cloud Firestore và Dataflow. Hoạt động sử dụng Dataflow sẽ được lập hoá đơn cho các tài nguyên mà các công việc của bạn sử dụng. Hãy xem trang đặt giá của Dataflow để biết thông tin chi tiết. Để biết giá của Cloud Firestore, hãy xem trang Giá.

Bước tiếp theo