تقدّم هذه الصفحة أمثلة حول كيفية استخدام Dataflow لإجراء عمليات مجمّعة في Cloud Firestore في مسار Apache. يتوافق شعاع Apache مع موصِّل لنظام Cloud Firestore. يمكنك استخدام هذا الموصل لتشغيل العمليات المجمّعة والبث في Dataflow.
ننصحك باستخدام Dataflow وApache Beam لمعالجة أحمال العمل المتعلقة بمعالجة البيانات على نطاق واسع.
يتوفّر موصل Cloud Firestore لـ Apache Beam بلغة Java. لمزيد من المعلومات حول موصل Cloud Firestore، يُرجى الاطّلاع على حزمة تطوير البرامج (SDK) لخدمة Apache Beam لـ Java.
قبل البدء
قبل قراءة هذه الصفحة، يجب أن تطّلع على نموذج برمجة شعاع Apache.
لتشغيل النماذج، يجب تفعيل واجهة برمجة التطبيقات 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; } }
يستخدم المثال الوسيطات التالية لإعداد مسار معيّن وتشغيله:
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 في Dataflow إلى تحمُّل تكاليف استخدام Cloud Firestore واستخدام Dataflow. وتتم الفوترة مقابل استخدام "تدفق البيانات" مقابل الموارد التي تستخدمها وظائفك. راجِع صفحة أسعار Dataflow لمعرفة التفاصيل. لمعرفة أسعار Cloud Firestore، يُرجى الاطّلاع على صفحة الأسعار.
الخطوات التالية
- راجع استخدام Firestore وApache Beam لمعالجة البيانات للحصول على مثال آخر لمسار.
- لمعرفة مزيد من المعلومات عن Dataflow وشعاع Apache، يُرجى الاطّلاع على مستندات Dataflow.