Bu sayfada, performans odaklı pazarlama için Dataflow'un Apache Beam ardışık düzeninde toplu Cloud Firestore işlemleri yapma. Apache Beam, Cloud Firestore için bir bağlayıcıyı destekler. Bunu kullanabilirsiniz Dataflow'da toplu işlem ve akış işlemlerini çalıştırmak için bağlayıcı.
Büyük ölçekli veriler için Dataflow ve Apache Beam kullanılmasını öneririz daha fazla bilgi edineceksiniz.
Apache Beam için Cloud Firestore bağlayıcısı Java'da kullanılabilir. Daha fazla Cloud Firestore bağlayıcısı hakkında bilgi edinmek için Java için Apache Beam SDK'sı.
Başlamadan önce
Bu sayfayı okumadan önce Apache Beam için programlama modeli.
Örnekleri çalıştırmak için Dataflow API'yi etkinleştirmeniz gerekir.Örnek Cloud Firestore ardışık düzenleri
Aşağıdaki örneklerde, veri yazan bir ardışık düzen gösterilmektedir. verileri okur ve filtreler. Bu örnekleri projeniz için başlangıç noktası olarak kullanabilirsiniz geliştirmenizi sağlar.
Örnek ardışık düzenlerini çalıştırma
Örneklerin kaynak kodu şu konumda bulunmaktadır: googleapis/java-firestore GitHub deposu. Bu örnekleri çalıştırmak için kaynak kodunu indirin ve BENİOKU'u inceleyin.
Örnek Write
ardışık düzeni
Aşağıdaki örnek, cities-beam-sample
koleksiyonunda dokümanlar oluşturur:
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; } }
Örnekte bir ardışık düzeni yapılandırmak ve çalıştırmak için aşağıdaki bağımsız değişkenler kullanılmaktadır:
GOOGLE_CLOUD_PROJECT=project-id REGION=region TEMP_LOCATION=gs://temp-bucket/temp/ NUM_WORKERS=number-workers MAX_NUM_WORKERS=max-number-workers
Örnek Read
Ardışık Düzeni
Aşağıdaki örnek ardışık düzen, cities-beam-sample
öğesindeki belgeleri okur
koleksiyonu, country
alanının şu şekilde ayarlandığı dokümanlar için filtre uygular:
USA
ve eşleşen dokümanların adlarını döndürür.
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); } })); } } }
Örnekte bir ardışık düzeni yapılandırmak ve çalıştırmak için aşağıdaki bağımsız değişkenler kullanılmaktadır:
GOOGLE_CLOUD_PROJECT=project-id REGION=region TEMP_LOCATION=gs://temp-bucket/temp/ NUM_WORKERS=number-workers MAX_NUM_WORKERS=max-number-workers
Fiyatlandırma
Dataflow'da Cloud Firestore iş yükü çalıştırmanın maliyeti vardır Cloud Firestore kullanımı ve Dataflow kullanımı için geçerlidir. Veri Akışı kullanımı, işlerinizin kullandığı kaynaklar için faturalandırılır. Bkz. Dataflow fiyatlandırma sayfası inceleyebilirsiniz. Cloud Firestore fiyatlandırması için bkz. Fiyatlandırma sayfası.
Sırada ne var?
- Başka bir ardışık düzen örneği için Veri işleme için Firestore ve Apache Beam'i kullanma bölümüne bakın.
- Dataflow ve Apache Beam hakkında daha fazla bilgi için Dataflow belgelerine bakın.