デベロッパー

Firestore と Apache Beam を使ったデータ処理

2021年11月18日

https://storage.googleapis.com/gweb-cloudblog-publish/images/firestore_header.max-1000x1000.jpeg

Google Cloud Japan Team

※この投稿は米国時間 2021 年 11 月 11 日に、Google Cloud blog に投稿されたものの抄訳です。

大規模なデータ処理ワークロードは、運用化とオーケストレーションが難しい場合があります。このたび、Google Cloud は、Apache Beam 用のネイティブモードの Firestore コネクタをリリースしました。Firestore ユーザーにとって、データ処理がこれまで以上に容易になりました。Apache Beam は、バッチ処理ストリーミング処理統合モデルで大規模なデータ処理を支援する、人気の高いオープンソースプロジェクトです。 Beam はポータビリティを備え、多種多様なバックエンドランナーと連携し、柔軟なデプロイが可能です。Firestore Beam I/O コネクタは、BigQuery、Bigtable、Datastore を Google データベースとして、Apache Beam コネクタと結合します。これは、Apache Beam Java SDK の Google Cloud Platform IO モジュールに自動的に含まれます。

Firestore コネクタは、Google Cloud Dataflow などのさまざまな Apache Beam バックエンドと併用できます。Dataflow は、Apache Beam のバックエンドランナーであり、デベロッパーが「驚異的並列ワークロード」の問題を解決する仕組みを提供します。データベースの各レコードを変更することは、このような問題の一例といえます。Beam パイプラインを使用することで、並列化のオーケストレーションの負担を減らし、デベロッパーはデータの変換に集中できます。

Beam の Firestore コネクタの実用例

Beam と Firestore パイプラインのユースケースをより深く理解するために、Google Cloud Dataflow を使用して Firestore データベースに一括処理を行うメリットの例を見てみましょう。例えば、コレクショングループ内のすべてのドキュメントを削除するなど、膨大な数の操作を行いたい Firestore データベースのコレクショングループがあるとします。これを 1 つのワーカーで行うと非常に時間がかかります。Beam を使って同時並行でできるとしたらどうでしょう？

読み込んでいます...

Pipeline pipeline = Pipeline.create(options);

String collectionGroupId = "collection-group-name";
RpcQosOptions rpcQosOptions = RpcQosOptions.newBuilder()
    .withHintMaxNumWorkers(options.as(DataflowPipelineOptions.class)
    .getMaxNumWorkers())
    .build();

pipeline
       .apply(Create.of(collectionGroupId))
       .apply(new CreatePartitionQueryRequest(rpcQosOptions.getHintMaxNumWorkers()))
       .apply(FirestoreIO.v1().read().partitionQuery().withNameOnlyQuery().build())
       .apply(FirestoreIO.v1().read().runQuery().build())
       .apply(MapElements.into(TypeDescriptors.strings()).via(
           (runQueryResponse) -> runQueryResponse.getDocument().getName())
       )
       .apply(ParDo.of(new CreateDeleteOperation()))
       .apply("shuffle writes", Reshuffle.viaRandomKey())
       .apply(
               FirestoreIO.v1().write()
                       .batchWrite()
                       .withRpcQosOptions(rpcQosOptions)
                       .build()
       );

pipeline.run().waitUntilFinish();

このパイプラインは、指定された collectionGroupId に対するパーティションクエリのリクエストを作成することから始まります。ネットワークの帯域幅の節減のため、withNameOnlyQuery を指定します。名前があればドキュメントを削除できます。いくつかカスタム関数を使用します。クエリのレスポンスを Document オブジェクトに読み込み、ドキュメント名を取得します。この名前でドキュメントを削除します。

読み込んでいます...

Beam は透かしを使って、正確に 1 回の処理を保証します。結果的に、Shuffle オペレーションでは、完了した作業の後戻りを防ぐことで、スピードと正確性を両立させています。

パーティションクエリを作成するコードは少し長いですが、生成された protobuf ビルダーを使用して、Firestore に送信する protobuf リクエストをビルドすることで構成されています。

パーティションクエリの作成

読み込んでいます...

static final class CreatePartitionQueryRequest extends
       PTransform<PCollection<String>, PCollection<PartitionQueryRequest>> {

private final long partitionCount;

public CreatePartitionQueryRequest(long partitionCount) {
       this.partitionCount = partitionCount;
   }

@Override
   public PCollection<PartitionQueryRequest> expand(PCollection<String> input) {
       return input.apply("create queries", ParDo.of(new DoFn<String, PartitionQueryRequest>() {
           @ProcessElement
           public void processElement(ProcessContext ctx) {
               String collectionId = ctx.element();
               String project = ctx.getPipelineOptions().as(GcpOptions.class).getProject();
               DatabaseRootName db = DatabaseRootName.of(project, "(default)");
               StructuredQuery.Builder query = StructuredQuery.newBuilder()
                       .addFrom(
                               CollectionSelector.newBuilder()
                                       .setCollectionId(collectionId)
                                       .setAllDescendants(true)
                       )
                       .addOrderBy(
                               Order.newBuilder()
                                     .setField(FieldReference.newBuilder().setFieldPath("__name__").build())
                                       .setDirection(Direction.ASCENDING)
                                       .build()
                       );
               PartitionQueryRequest request = PartitionQueryRequest.newBuilder()
                       .setParent(db.toString() + "/documents")
                       .setPartitionCount(partitionCount - 1)
                       .setStructuredQuery(query)
                       .build();
               ctx.output(request);
           }
       }));
   }
}

このコネクタは、Google Cloud のユーザーにとってさまざまな用途が考えられます。ネイティブモードの Firestore データベースでの異種データの結合、複数のデータベースのデータ関連付け、大量のエンティティの削除、Firestore データの BigQuery への書き込みなどです。このコネクタを Apache Beam エコシステムに提供できたことをうれしく思います。また、皆さまが Firestore コネクタをどのように使用して次の素晴らしいサービスを構築するか楽しみです。

- Google Cloud スタッフエンジニア、Chris Wilcox

- Google Cloud Developer Relations エンジニア、Ben Whitehead

データベース