داده ها را به صورت انبوه با Dataflow پردازش کنید

این صفحه نمونه‌هایی از نحوه استفاده از Dataflow برای انجام عملیات انبوه Cloud Firestore در یک خط لوله Apache Beam را ارائه می‌دهد. Apache Beam از یک کانکتور برای Cloud Firestore پشتیبانی می‌کند. می‌توانید از این کانکتور برای اجرای عملیات دسته‌ای و جریانی در Dataflow استفاده کنید.

ما استفاده از Dataflow و Apache Beam را برای حجم کاری پردازش داده‌های بزرگ توصیه می‌کنیم.

کانکتور Cloud Firestore برای Apache Beam در جاوا موجود است. برای اطلاعات بیشتر در مورد کانکتور Cloud Firestore ، به Apache Beam SDK برای جاوا مراجعه کنید.

قبل از اینکه شروع کنی

قبل از اینکه این صفحه را بخوانید، باید با مدل برنامه‌نویسی آپاچی بیم آشنا باشید.

برای اجرای نمونه‌ها، باید Dataflow API را فعال کنید .

مثالی از خطوط لوله Cloud Firestore

مثال‌های زیر یک خط لوله (pipeline) را نشان می‌دهند که داده‌ها را می‌نویسد و دیگری داده‌ها را می‌خواند و فیلتر می‌کند. می‌توانید از این نمونه‌ها به عنوان نقطه شروع برای خطوط لوله خودتان استفاده کنید.

اجرای خطوط لوله نمونه

کد منبع نمونه‌ها در مخزن گیت‌هاب googleapis/java-firestore موجود است. برای اجرای این نمونه‌ها، کد منبع را دانلود کرده و فایل 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

این مثال از آرگومان‌های زیر برای پیکربندی و اجرای یک pipeline استفاده می‌کند:

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

این مثال از آرگومان‌های زیر برای پیکربندی و اجرای یک pipeline استفاده می‌کند:

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 به همراه دارد. هزینه استفاده از Dataflow برای منابعی که کارهای شما استفاده می‌کنند، محاسبه می‌شود. برای جزئیات بیشتر به صفحه قیمت‌گذاری Dataflow مراجعه کنید. برای قیمت‌گذاری Cloud Firestore ، به صفحه قیمت‌گذاری مراجعه کنید.

قدم بعدی چیست؟