इस पेज पर उदाहरण के तौर पर, बेहतर परफ़ॉर्म करने के लिए Dataflow का इस्तेमाल किया गया है Apache बीम पाइपलाइन में बल्क Cloud Firestore ऑपरेशन. Apache बीम, Cloud Firestore के कनेक्टर के साथ काम करता है. आप इसका इस्तेमाल कर सकते हैं Dataflow में बैच और स्ट्रीमिंग से जुड़ी कार्रवाइयां चलाने के लिए कनेक्टर.
बड़े पैमाने पर डेटा इकट्ठा करने के लिए, हम Dataflow और Apache बीम का इस्तेमाल करने का सुझाव देते हैं वर्कलोड को प्रोसेस किया जा रहा है.
Apache बीम के लिए Cloud Firestore कनेक्टर, Java में उपलब्ध है. ज़्यादा के लिए Cloud Firestore कनेक्टर के बारे में जानकारी देखने के लिए, Java के लिए Apache बीम SDK टूल.
शुरू करने से पहले
इस पेज को पढ़ने से पहले, आपको Apache बीम के लिए प्रोग्रामिंग मॉडल.
सैंपल चलाने के लिए, आपको 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
कीमत
Dataflow में Cloud Firestore वर्कलोड चलाने पर लागत आती है Cloud Firestore के इस्तेमाल और Dataflow के इस्तेमाल के लिए. डेटाफ़्लो इस्तेमाल का बिल उन संसाधनों के आधार पर लिया जाता है जिनका इस्तेमाल आपकी जॉब में किया जाता है. ज़्यादा जानकारी के लिए, Dataflow की कीमत का पेज देखें. Cloud Firestore की कीमतों के बारे में जानने के लिए, यहां देखें कीमत की जानकारी देने वाला पेज.
आगे क्या करना है
- पाइपलाइन के दूसरे उदाहरण के लिए, डेटा प्रोसेसिंग के लिए Firestore और Apache बीम का इस्तेमाल करना लेख पढ़ें.
- Dataflow और Apache बीम के बारे में ज़्यादा जानने के लिए, Dataflow का दस्तावेज़ देखें.