Traiter des données de manière groupée avec Dataflow
Cette page donne des exemples d'utilisation de Dataflow pour effectuer des opérations Firestore groupées dans un pipeline Apache Beam. Apache Beam est compatible avec un connecteur pour Firestore. Vous pouvez utiliser ce connecteur pour exécuter des opérations par lot et par flux dans Dataflow.
Nous vous recommandons d'utiliser Dataflow et Apache Beam pour les charges de travail de traitement de données à grande échelle.
Le connecteur Firestore pour Apache Beam est disponible en Java. Pour en savoir plus sur le connecteur Firestore, consultez la page SDK Apache Beam pour Java.
Avant de commencer
Avant de lire cette page, vous devez avoir étudié le modèle de programmation pour Apache Beam.
Pour exécuter les exemples, vous devez activer l'API Dataflow:
Exemples de pipelines Firestore
Les exemples ci-dessous illustrent un pipeline qui écrit des données, et un autre qui lit et filtre des données. Vous pouvez utiliser ces exemples comme point de départ pour vos propres pipelines.
Exécuter les exemples de pipelines
Le code source des exemples est disponible dans le dépôt GitHub googleapis/java-firestore. Pour exécuter ces exemples, téléchargez le code source et consultez le fichier README.
Exemple de pipeline Write
L'exemple suivant crée des documents dans la collection 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; } }
L'exemple utilise les arguments suivants pour configurer et exécuter un pipeline:
GOOGLE_CLOUD_PROJECT=project-id REGION=region TEMP_LOCATION=gs://temp-bucket/temp/ NUM_WORKERS=number-workers MAX_NUM_WORKERS=max-number-workers
Exemple de pipeline Read
L'exemple de pipeline suivant lit des documents de la collection cities-beam-sample
, applique un filtre aux documents où le champ country
est défini sur USA
et renvoie les noms des documents correspondants.
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); } })); } } }
L'exemple utilise les arguments suivants pour configurer et exécuter un pipeline:
GOOGLE_CLOUD_PROJECT=project-id REGION=region TEMP_LOCATION=gs://temp-bucket/temp/ NUM_WORKERS=number-workers MAX_NUM_WORKERS=max-number-workers
Tarification
L'exécution d'une charge de travail Firestore dans Dataflow entraîne des coûts d'utilisation de Firestore et de Dataflow. L'utilisation de Dataflow est facturée en fonction des ressources utilisées par vos jobs. Consultez la page des tarifs de Dataflow pour en savoir plus. Pour en savoir plus sur les tarifs de Firestore, consultez la page des tarifs.
Étapes suivantes
- Pour obtenir un autre exemple de pipeline, consultez la page Utiliser Firestore et Apache Beam pour le traitement des données.
- Pour en savoir plus sur Dataflow et Apache Beam, consultez la documentation Dataflow.