এই পৃষ্ঠাটি একটি Apache Beam পাইপলাইনে বাল্ক Cloud Firestore অপারেশনগুলি সম্পাদন করতে ডেটাফ্লো ব্যবহার করার উদাহরণ দেয়৷ Apache Beam Cloud Firestore এর জন্য একটি সংযোগকারীকে সমর্থন করে। আপনি ডেটাফ্লোতে ব্যাচ এবং স্ট্রিমিং অপারেশন চালানোর জন্য এই সংযোগকারীটি ব্যবহার করতে পারেন।
আমরা বড় আকারের ডেটা প্রসেসিং ওয়ার্কলোডের জন্য Dataflow এবং Apache Beam ব্যবহার করার পরামর্শ দিই।
Apache Beam-এর জন্য Cloud Firestore সংযোগকারী জাভাতে উপলব্ধ। Cloud Firestore সংযোগকারী সম্পর্কে আরও তথ্যের জন্য, Java এর জন্য Apache Beam SDK দেখুন।
আপনি শুরু করার আগে
আপনি এই পৃষ্ঠাটি পড়ার আগে, আপনার Apache Beam-এর জন্য প্রোগ্রামিং মডেলের সাথে পরিচিত হওয়া উচিত।
নমুনা চালানোর জন্য, আপনাকে অবশ্যই Dataflow API সক্ষম করতে হবে।উদাহরণ Cloud Firestore পাইপলাইন
নীচের উদাহরণগুলি একটি পাইপলাইন প্রদর্শন করে যা ডেটা লেখে এবং একটি যা ডেটা পড়ে এবং ফিল্টার করে৷ আপনি এই নমুনাগুলি আপনার নিজের পাইপলাইনের জন্য একটি সূচনা পয়েন্ট হিসাবে ব্যবহার করতে পারেন।
নমুনা পাইপলাইন চলমান
নমুনার সোর্স কোড googleapis/java-firestore GitHub সংগ্রহস্থলে পাওয়া যায়। এই নমুনাগুলি চালানোর জন্য, সোর্স কোডটি ডাউনলোড করুন এবং 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;
}
}উদাহরণটি একটি পাইপলাইন কনফিগার এবং চালানোর জন্য নিম্নলিখিত আর্গুমেন্টগুলি ব্যবহার করে:
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);
}
}));
}
}
}উদাহরণটি একটি পাইপলাইন কনফিগার এবং চালানোর জন্য নিম্নলিখিত আর্গুমেন্টগুলি ব্যবহার করে:
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 ব্যবহার এবং ডেটাফ্লো ব্যবহারের জন্য খরচ হয়৷ ডেটাফ্লো ব্যবহার আপনার কাজগুলি ব্যবহার করে এমন সংস্থানগুলির জন্য বিল করা হয়। বিস্তারিত জানার জন্য Dataflow মূল্য পৃষ্ঠা দেখুন। Cloud Firestore মূল্যের জন্য, মূল্য নির্ধারণের পৃষ্ঠাটি দেখুন।
এরপর কি
- আরেকটি পাইপলাইনের উদাহরণের জন্য ডেটা প্রক্রিয়াকরণের জন্য Firestore এবং Apache Beam ব্যবহার করা দেখুন।
- Dataflow এবং Apache Beam সম্পর্কে আরও জানতে, Dataflow ডকুমেন্টেশন দেখুন।