Dataflow ile verileri toplu olarak işleyin

Bu sayfada, Apache Beam işlem hattında toplu Cloud Firestore işlemlerini gerçekleştirmek için Dataflow'un nasıl kullanılacağına ilişkin örnekler verilmektedir. Apache Beam, Cloud Firestore için bir bağlayıcıyı destekler. Dataflow'da toplu ve akış işlemlerini çalıştırmak için bu bağlayıcıyı kullanabilirsiniz.

Büyük ölçekli veri işleme iş yükleri için Dataflow ve Apache Beam'i kullanmanızı öneririz.

Apache Beam için Cloud Firestore bağlayıcısı Java'da mevcuttur. Cloud Firestore bağlayıcısı hakkında daha fazla bilgi için Java için Apache Beam SDK'sına bakın.

Sen başlamadan önce

Bu sayfayı okumadan önce Apache Beam için Programlama modeline aşina olmalısınız.

Örnekleri çalıştırmak için Dataflow API'yi etkinleştirmeniz gerekir.

Örnek Cloud Firestore işlem hatları

Aşağıdaki örneklerde veri yazan ve verileri okuyan ve filtreleyen bir işlem hattı gösterilmektedir. Bu örnekleri kendi işlem hatlarınız için bir başlangıç ​​noktası olarak kullanabilirsiniz.

Örnek işlem hatlarını çalıştırma

Örneklerin kaynak kodu googleapis/java-firestore GitHub deposunda mevcuttur. Bu örnekleri çalıştırmak için kaynak kodunu indirin ve README dosyasına bakın.

Örnek Write hattı

Aşağıdaki örnek, cities-beam-sample koleksiyonunda belgeler oluşturur:



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

Örnek, bir işlem hattını yapılandırmak ve çalıştırmak için aşağıdaki bağımsız değişkenleri kullanır:

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

Örnek Okuma Read Düzeni

Aşağıdaki örnek işlem hattı, cities-beam-sample koleksiyonundaki belgeleri okur, country alanının USA olarak ayarlandığı belgelere bir filtre uygular ve eşleşen belgelerin adlarını döndürür.



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

Örnek, bir işlem hattını yapılandırmak ve çalıştırmak için aşağıdaki bağımsız değişkenleri kullanır:

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

Fiyatlandırma

Dataflow'da Cloud Firestore iş yükünü çalıştırmak, Cloud Firestore kullanımı ve Dataflow kullanımına ilişkin maliyetler doğurur. Veri akışı kullanımı, işlerinizin kullandığı kaynaklar için faturalandırılır. Ayrıntılar için Dataflow fiyatlandırma sayfasına bakın. Cloud Firestore fiyatlandırması için Fiyatlandırma sayfasına bakın.

Sıradaki ne