หน้านี้จะแสดงตัวอย่างวิธีใช้ Dataflow เพื่อดำเนินการ Cloud Firestore จำนวนมากในไปป์ไลน์ Apacheบีม Apache Beam รองรับเครื่องมือเชื่อมต่อสำหรับ Cloud Firestore คุณใช้เครื่องมือเชื่อมต่อนี้เพื่อเรียกใช้การดำเนินการแบบกลุ่มและสตรีมมิงใน Dataflow ได้
เราขอแนะนำให้ใช้ Dataflow และ Apache Beam สำหรับภาระงานการประมวลผลข้อมูลขนาดใหญ่
เครื่องมือเชื่อมต่อ Cloud Firestore สำหรับ Apache Beam จะมีให้ใช้งานใน Java ดูข้อมูลเพิ่มเติมเกี่ยวกับเครื่องมือเชื่อมต่อ Cloud Firestore ได้ที่ Apache Beam SDK สำหรับ Java
ก่อนเริ่มต้น
ก่อนที่จะอ่านหน้านี้ คุณควรทำความคุ้นเคยกับ โมเดลการเขียนโปรแกรมสำหรับ 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
ราคา
การเรียกใช้ภาระงาน Cloud Firestore ใน Dataflow ทำให้เกิดค่าใช้จ่ายสำหรับการใช้ Cloud Firestore และการใช้งาน Dataflow ระบบจะเรียกเก็บเงินการใช้งาน Dataflow สำหรับทรัพยากรที่งานของคุณใช้ ดูรายละเอียดในหน้าการกำหนดราคาของ Dataflow โปรดดูราคาของ Cloud Firestore ที่หน้าราคา
ขั้นตอนถัดไป
- ดูการใช้ Firestore และ Apache Beam สำหรับการประมวลผลข้อมูลสำหรับตัวอย่างไปป์ไลน์อื่น
- ดูข้อมูลเพิ่มเติมเกี่ยวกับ Dataflow และ Apache Beam ได้ในเอกสารประกอบเกี่ยวกับ Dataflow