עיבוד נתונים בכמות גדולה באמצעות Dataflow

בדף הזה מפורטות דוגמאות לשימוש ב-Dataflow כדי לבצע פעולות Cloud Firestore בכמות גדולה בצינור עיבוד נתונים של Apache Beam. ב-Apache Beam יש מחבר ל-Cloud Firestore. אפשר להשתמש במחבר הזה כדי להריץ פעולות באצווה ובסטרימינג ב-Dataflow.

מומלץ להשתמש ב-Dataflow וב-Apache Beam לעומסי עבודה (workloads) גדולים של עיבוד נתונים.

המחבר Cloud Firestore ל-Apache Beam זמין ב-Java. מידע נוסף על המחבר Cloud Firestore זמין במאמר Apache Beam SDK for Java.

לפני שמתחילים

לפני שקוראים את הדף הזה, כדאי להכיר את מודל התכנות של Apache Beam.

כדי להריץ את הדוגמאות, צריך להפעיל את Dataflow API.

צינורות עיבוד נתונים לדוגמה של Cloud Firestore

הדוגמאות הבאות מדגימות צינור עיבוד נתונים שכותב נתונים וצינור עיבוד נתונים שקורא ומסנן נתונים. אפשר להשתמש בדוגמאות האלה כנקודת התחלה לצינורות עיבוד נתונים משלכם.

הפעלת צינורות עיבוד הנתונים לדוגמה

קוד המקור של הדוגמאות זמין במאגר GitHub‏ 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

בדוגמה הזו נעשה שימוש בארגומנטים הבאים כדי להגדיר ולהפעיל צינור עיבוד נתונים:

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. החיוב על השימוש ב-Dataflow מתייחס למשאבים שבהם משתמשות המשימות. פרטים נוספים זמינים בדף התמחור של Dataflow. למידע על התמחור של Cloud Firestore, ראו דף התמחור.

המאמרים הבאים

• דוגמה נוספת לצינור עיבוד נתונים מופיעה במאמר שימוש ב-Firestore וב-Apache Beam לעיבוד נתונים.
• מידע נוסף על Dataflow ו-Apache Beam זמין במסמכי העזרה של Dataflow.