ডাটাফ্লো দিয়ে প্রচুর পরিমাণে ডেটা প্রসেস করুন

এই পৃষ্ঠায় অ্যাপাচি বিম পাইপলাইনে ডেটাফ্লো ব্যবহার করে কীভাবে বাল্ক Cloud Firestore অপারেশন সম্পাদন করা যায় তার উদাহরণ দেওয়া হয়েছে। অ্যাপাচি বিম Cloud Firestore জন্য একটি কানেক্টর সমর্থন করে। আপনি ডেটাফ্লোতে ব্যাচ এবং স্ট্রিমিং অপারেশন চালানোর জন্য এই কানেক্টরটি ব্যবহার করতে পারেন।

বৃহৎ পরিসরের ডেটা প্রসেসিং কাজের জন্য আমরা ডেটাফ্লো এবং অ্যাপাচি বিম ব্যবহারের পরামর্শ দিই।

অ্যাপাচি বিম-এর জন্য Cloud Firestore কানেক্টরটি জাভাতে উপলব্ধ। Cloud Firestore কানেক্টর সম্পর্কে আরও তথ্যের জন্য, জাভার জন্য অ্যাপাচি বিম এসডিকে (Sapache Beam SDK for Java ) দেখুন।

শুরু করার আগে

এই পৃষ্ঠাটি পড়ার আগে, আপনার অ্যাপাচি বিম (Apache Beam)-এর প্রোগ্রামিং মডেল সম্পর্কে পরিচিত থাকা উচিত।

স্যাম্পলগুলো চালানোর জন্য, আপনাকে ডেটাফ্লো এপিআই (Dataflow API) সক্রিয় করতে হবে।

Cloud Firestore পাইপলাইনের উদাহরণ

নিচের উদাহরণগুলো ডেটা লেখার এবং ডেটা পড়া ও ফিল্টার করার একটি পাইপলাইন প্রদর্শন করে। আপনি আপনার নিজস্ব পাইপলাইনের জন্য এই নমুনাগুলোকে প্রাথমিক ভিত্তি হিসেবে ব্যবহার করতে পারেন।

নমুনা পাইপলাইনগুলি চালানো হচ্ছে

স্যাম্পলগুলোর সোর্স কোড 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 ওয়ার্কলোড চালালে Cloud Firestore এবং ডেটাফ্লো ব্যবহারের জন্য খরচ হয়। আপনার জবগুলো যে রিসোর্স ব্যবহার করে, তার জন্যই ডেটাফ্লো ব্যবহারের বিল করা হয়। বিস্তারিত জানতে ডেটাফ্লো প্রাইসিং পেজ দেখুন। Cloud Firestore প্রাইসিং জানতে প্রাইসিং পেজ দেখুন।

এরপর কী?

• আরেকটি পাইপলাইন উদাহরণের জন্য ‘ডেটা প্রসেসিংয়ের জন্য ফায়ারস্টোর এবং অ্যাপাচি বিম ব্যবহার’ দেখুন।
• Dataflow এবং Apache Beam সম্পর্কে আরও জানতে, Dataflow ডকুমেন্টেশন দেখুন।