Write a stream into Firestore#

Assumes the artifacts and credentials from the Quickstart index, and the imports an IDE resolves from the Java API reference.

Create the database first. The sink writes to an existing Firestore database in Native mode and creates no database; collections need no creating. A project’s (default) database is created once, for example with:

gcloud firestore databases create --location=asia-northeast1

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
// Not optional: the sink is at-least-once only with checkpointing, which is what makes
// Flink wait for every write to be answered before the barrier passes.
env.enableCheckpointing(60_000);

env.fromData("a-1", "a-2")
        .sinkTo(
                FirestoreSink.<String>builder()
                        .database(DatabaseDestination.of("my-project"))
                        .serializer(
                                (element, context) ->
                                        // set, not create: the sink is at-least-once, so
                                        // a record can arrive twice. A set makes that a
                                        // no-op; a create makes it a routed failure.
                                        FirestoreWrite.set(
                                                "orders/" + element,
                                                Map.of("total", (long) element.length())))
                        .build());

env.execute("firestore-quickstart");

The documents appear under the orders collection in the Firestore console, or read them back with the client library.

Two things are decided in that job rather than by the sink. The document comes from the write, not from the builder: one sink writes to as many collections as the serializer names. And the operation is what decides whether a replay is harmless; the connector page has the table of which operations are idempotent.

The identity running the job needs permission to create, update and delete documents, which roles/datastore.user carries.

Next#

Firestore examples covers merges, deletes, conditional updates, dropping refused writes, and local development against the emulator.