Firestore examples#

Worked cases for the Firestore sink. The basic job is the Quickstart; what each option does is on the connector page.

DataStream sink#

Merging fields into existing documents#

setMerge writes the fields it names and leaves the others in place, creating the document if it is missing. A nested map is merged key by key, so the example updates one key of audit without replacing the map.

FirestoreSink.<OrderEvent>builder()
        .database(DatabaseDestination.of("my-project"))
        .serializer(
                (event, context) ->
                        FirestoreWrite.setMerge(
                                "orders/" + event.getId(),
                                Map.of(
                                        "status", event.getStatus(),
                                        // A nested map is merged key by key, so other
                                        // keys of "audit" keep their values.
                                        "audit", Map.of("lastStatus", event.getStatus()))))
        .build();

update also leaves unnamed fields alone, but it replaces a named map as a whole and fails the job when the document is missing; prefer setMerge when a document may not exist yet.

Deleting documents#

A serializer can choose the operation per record. Deleting a missing document succeeds, so a replayed delete is harmless.

FirestoreSink.<OrderEvent>builder()
        .database(DatabaseDestination.of("my-project"))
        .serializer(
                (event, context) -> {
                    String path = "orders/" + event.getId();
                    return event.isCancelled()
                            ? FirestoreWrite.delete(path)
                            : FirestoreWrite.set(path, Map.of("total", event.getTotal()));
                })
        .build();

Writes to the same document are not ordered: a set and a later delete of one document may be applied in either order. A stream that deletes and rewrites the same documents in quick succession needs an order the sink does not keep yet (#1556).

Updating a document only if it has not changed#

An update or delete given a lastUpdateTime applies only if the document was last updated at exactly that time. Here a price change carries the getUpdateTime() of the document snapshot it was computed from, so it applies only if nothing changed the document since that snapshot; a snapshot’s read time would not do, because the precondition compares the document’s last update time exactly. A write that lost the race is refused with FAILED_PRECONDITION, and the routing policy hands it to the failure handler instead of failing the job.

FirestoreSink.<PriceChange>builder()
        .database(DatabaseDestination.of("my-project"))
        .serializer(
                (change, context) ->
                        FirestoreWrite.update(
                                "products/" + change.getProductId(),
                                Map.of("price", change.getPrice()),
                                change.getUpdateTime()))
        .preconditionFailurePolicy(PreconditionFailurePolicy.ROUTE_TO_FAILURE_HANDLER)
        .failedWriteHandler(FailureHandler.logAndDrop())
        .build();

A replay of a conditional write that had already been applied is refused the same way, because its first application changed the update time. Under the default policy that fails the job again on every restart, which is why a stream of conditional writes usually wants this policy.

Dropping refused writes instead of failing the job#

A dropping handler keeps the job running through writes Firestore refuses, such as a document over 1 MiB. maxConsecutiveRejections fails the job anyway once that many refusals arrive with nothing applied between them, which is what a stream whose data is broken wholesale looks like.

FirestoreSink.<OrderEvent>builder()
        .database(DatabaseDestination.of("my-project"))
        .serializer(orderSerializer)
        .failedWriteHandler(FailureHandler.logAndDrop())
        .writerOptions(
                FirestoreWriterOptions.builder()
                        // Fail after 20 confirmed refusals with none applied between.
                        .maxConsecutiveRejections(20)
                        .build())
        .build();

Local development#

Running against the emulator#

Start the emulator from the Google Cloud CLI:

gcloud emulators firestore start --host-port=localhost:8080

Then point the sink at it. The emulator needs no credentials, so the sink sends none.

FirestoreSink.<OrderEvent>builder()
        .database(DatabaseDestination.of("my-project"))
        .serializer(orderSerializer)
        .emulatorEndpoint("localhost:8080")
        .build();

Set the endpoint on the builder rather than exporting FIRESTORE_EMULATOR_HOST. The client library would honor the variable too, and would then send a job meant for the real service to the emulator; the sink warns when the variable is set without an endpoint. The emulator is not the service: it does not enforce the request-size limit or IAM, and the deviation table lists what is known to differ.