Trang này cung cấp các ví dụ về cách sử dụng Dataflow để thực hiện các hoạt động Cloud Firestore hàng loạt trong đường dẫn Apache Beam. Apache Beam hỗ trợ trình kết nối cho Cloud Firestore. Bạn có thể sử dụng trình kết nối này để chạy các hoạt động hàng loạt và truyền phát trong Dataflow.
Chúng tôi khuyên bạn nên sử dụng Dataflow và Apache Beam cho khối lượng công việc xử lý dữ liệu quy mô lớn.
Trình kết nối Cloud Firestore cho Apache Beam có sẵn bằng Java. Để biết thêm thông tin về trình kết nối Cloud Firestore, hãy xem Apache Beam SDK cho Java .
Trước khi bắt đầu
Trước khi đọc trang này, bạn nên làm quen với Mô hình lập trình cho Apache Beam .
Để chạy mẫu, bạn phải bật API Dataflow .Ví dụ về đường ống Cloud Firestore
Các ví dụ bên dưới minh họa một quy trình ghi dữ liệu và một quy trình đọc và lọc dữ liệu. Bạn có thể sử dụng các mẫu này làm điểm khởi đầu cho quy trình của riêng mình.
Chạy đường ống mẫu
Mã nguồn của các mẫu có sẵn trong kho lưu trữ GitHub của googleapis/java-firestore . Để chạy các mẫu này, hãy tải xuống mã nguồn và xem README .
Ví dụ Write đường dẫn
Ví dụ sau đây tạo tài liệu trong bộ sưu tập 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;
}
}Ví dụ này sử dụng các đối số sau để đặt cấu hình và chạy quy trình:
GOOGLE_CLOUD_PROJECT=project-id REGION=region TEMP_LOCATION=gs://temp-bucket/temp/ NUM_WORKERS=number-workers MAX_NUM_WORKERS=max-number-workers
Đường dẫn Read ví dụ
Quy trình mẫu sau đây đọc tài liệu từ bộ sưu tập cities-beam-sample , áp dụng bộ lọc cho các tài liệu trong đó trường country được đặt thành USA và trả về tên của các tài liệu phù hợp.
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);
}
}));
}
}
}Ví dụ này sử dụng các đối số sau để đặt cấu hình và chạy quy trình:
GOOGLE_CLOUD_PROJECT=project-id REGION=region TEMP_LOCATION=gs://temp-bucket/temp/ NUM_WORKERS=number-workers MAX_NUM_WORKERS=max-number-workers
Định giá
Việc chạy khối lượng công việc của Cloud Firestore trong Dataflow sẽ phát sinh chi phí cho việc sử dụng Cloud Firestore và việc sử dụng Dataflow. Việc sử dụng luồng dữ liệu được tính phí cho các tài nguyên mà công việc của bạn sử dụng. Xem trang giá Dataflow để biết chi tiết. Để biết giá của Cloud Firestore, hãy xem trang Định giá .
Cái gì tiếp theo
- Xem Sử dụng Firestore và Apache Beam để xử lý dữ liệu để biết ví dụ về đường dẫn khác.
- Để biết thêm về Dataflow và Apache Beam, hãy xem tài liệu Dataflow .