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