AWSエンジニアがGoogle Cloudに挑む：Iceberg×Geminiで実現するプラットフォーム構築の舞台裏

はじめに

NTTドコモデータプラットフォーム部外山です。

NTTドコモではお客様起点の事業運営を実現するために、その声を誰でも簡単に分析できるアプリを社内データ活用プラットフォームPochi*1上で開発・利用しています。

声の利活用においては、一つ一つのお客様の声データにデモグラ情報や、ポジティブ評価、ネガティブ評価を行い、非構造化データを分析可能にする必要があるため、過去に以下のブログ記事でご紹介した専用のデータパイプラインを構築して利用していました。

より様々なお客様の声データを⼀元的に取り扱えるよう、Google Cloud上に新たにプラットフォームを構築することとなりました。

今回その内容を解説しますが、プラットフォーム構築の検討・実装についてはNTTデータの根岸さんに進めてもらっており、以下については根岸さんに執筆いただいています。根岸さんはAWSの知見は深い一方、今回Google Cloudでのプラットフォーム構築を初めて行ったこともあり、構築を通じて感じた両クラウドサービス間での違いも併せてご紹介いたします。

社内データ活用プラットフォームPochiとは

私たちDP部は社内のデータ民主化を目指し、StreamlitとGoogle Cloudで 圧倒的に使いやすい データ活用プラットフォームを開発・推進しています。このプラットフォームは、24年度は30万時間もの業務効率化を実現し、直近では5,000人以上の社員に利用が拡大しています。

ASCII.jp：NTTドコモ、Streamlit利用の“ポチポチ分析アプリ”開発で社内データ活用を促進 (1/3)

プラットフォームとして目指す姿

要件をもとに以下3点の実現を目指すこととし、検討の結果、今回はGoogle Cloudを利用することとしました。

1. 社内外問わず、様々なデータを集約できること

現状、お客様の声データは社内の各システムに散在し、SNSで寄せられる声など社外データも存在します。それらのシステムはそれぞれ異なった形式でデータを蓄積しており、データ形式に問わずデータを格納できるようなサービスが必要でした。

今回はGoogle Cloud Storageを生データの格納場所とすることで、柔軟かつ可用性の高いデータレイクとして活用することにしました。また、Google Cloudが提供するBigQueryは完全にサーバレスなデータウェアハウスであり、幅広いデータを柔軟に格納できることに加え、ほとんど下準備無しに利用開始できるという点が、今回の案件で求められているスピード感にマッチしていました。

2.お客様の声データを分析しやすいように付加情報を手軽に付与できること

お客様の声データはフリーコメントであることから決まった形式を持たないため、分析を行うにはその文章の中から分析に適した情報を適切な形で取り出す必要があります。そのため、従来からデータ活用プラットフォームで行われているような正規表現によるデータクリーニングも必要ですが、それに加えてLLM処理を用いての内容要約や、声を類型化するためのエンベディング処理など、生成AIを用いたデータ加工が必須となります。

Google社が提供しているGoogle Geminiは最大約200万トークン級の長いコンテキストを扱うことができ、さらにテキスト・画像・動画・音声なども同時に処理できる（ネイティブ・マルチモーダル）ため、今後利用規模を拡張していくにあたって様々なデータに柔軟に対応していけるという点が決め手となりました。（他のクラウドサービスでもモデルの選択によってはそれぞれのユースケースで対応可能ではあるものの、Geminiでは同じモデルでも万能に対応できるという点がメリットとも考えています。）

また、今回のプラットフォームでは別の構成を採用していますがBigQuery × Vertex AIにおいては追加ミドルウェアなしでSQL内で直接モデルを呼び出すことができ、その点についてもデータ活用プラットフォームとしてBigQueryを用いるメリットのひとつだと考えています。

3.プラットフォームで加工したデータを全社的に活用してもらうこと

せっかく集約・加工したデータも幅広いユーザに活用してもらわないと意味がありません。

本プロジェクトで加工されたデータの主な活用先は社内データ活用プラットフォームPochiのアプリを想定しておりますが、他の利活用においてはSnowflake上に作成したデータを公開することや他システムへの連携（例えば、AWSで構築された配信システム）が必要となります。これまでこうしたシステム間でのやり取りは、加工済みの実体データを配置する前提で検討が必要でしたが、 OpenTableFormat（Iceberg）の登場により、システムを跨いだ透過的な活用が可能となっています。

上述の検討結果及び、社内データ活用プラットフォームPochiは以下ブログの通り、Google Cloudで構成されていることも踏まえて、今回のプラットフォームについてもGoogle Cloud上にで構築することとしました。

nttdocomo-developers.jp

プラットフォームの実装例

プラットフォームの構成概要図は以下の通りです。

Iceberg技術を用いて透過的なデータ連携を実現する。

今回の案件におけるプラットフォームの構成は、一般的なデータ統合のプロセスである ELT（ ①Extract(抽出)・②Load(格納)・③Transform(変換) ） の形になっています。 BigQueryとして得意な処理・Snowflakeとして得意な処理をそれぞれ役割分担することで効率良くデータを処理しつつも、前述のOpenTableFormat（Iceberg）の導入によりそれらの境界を意識させないような構成とすることで、Google Cloudから見て、外部システム（Snowflake）内の処理も含めた、一元管理されたパイプラインを実現しました。

個人的な感覚にはなりますが、AWSは細かい部品を組み合わせて最適解を設計していくというイメージに対して、Google Cloudは細かいことを気にせずにまずはサービスを体感してもらうというイメージを持っています。どちらかが優れているということは無く、そのプロジェクトの状況や要件などに左右される部分が大きいとは思いますが、今回のプロジェクトでは設計～リリースまでかなりのスピード感を求められていたこともあり、Google Cloudのまずは動かしてみるというスタンスにはかなり助けられました。

以下、⓪データパイプラインのオーケストレーションおよび ①Extract・②Load・③Transformのそれぞれの概要について記載していきます。

⓪ データパイプラインのオーケストレーション

データパイプラインはGoogle Cloudが提供しているCloud Composer 3（以下、Composer）を利用してオーケストレーションを行っています。

Composerは、Apache Airflowをベースに構築されたフルマネージドのワークフローオーケストレーションサービスであり、DAG（Directed Acyclic Graph）を用意することで、一連のタスクとその依存関係を表現することができます。このApache Airflowは実体はGoogle Kubernetes Engine（GKE）上で稼働しているのですが、そのGKEの管理や運用保守までまとめて実施してくれるのがGoogleが提供しているComposerを利用する大きなメリットとなっています。

AWSでも「Amazon Managed Workflows for Apache Airflow（MWAA）」という名称でAirflowをベースとしたオーケストレーションのサービスがあります。両者ともAirflowをベースとしているため大きな機能差はなく、基本的に自社で活用しているクラウドサービスに準じた活用を検討することで問題ありませんが、Composerでは、CelerKubernetesExecuterという機能を標準搭載しており、軽い処理はCeley・重い処理はKubernetesExecuterで実行するといった住み分けが可能である点がメリットとなります。

現時点でこのプロジェクトにおいては上記の機能は利用していませんが、今後パイプライン上で大量データを扱うような重い処理が追加されることが想定されており、今後活用予定となっています。

① Extract処理

今後様々なデータソースからこのプラットフォームに対してデータを取り込んでいく予定ですが、現時点ではSnowflakeに格納されているお客様の声データのみを取り込んでいます。

プラットフォーム観点では、まずGoogle Cloud StorageをSnowflakeの外部ストレージとして認識させ、Snowflake側でタスクを実行することによって外部ストレージを経由してデータの取り込みを実現しています。

タスク内ではまずSnowflake側でIcebergTableを作成しGET_ICEBERG_TABLE_INFORMATION関数を実行することで最新のメタデータファイルパスを取得、その取得したパスを外部ストレージ経由で受け渡し、BigQuery側でApache Iceberg external tableを作成するような動きをしています。

最新のメタデータファイルパスを取得するクエリ例

    query = f"""
        SELECT
            JSON_EXTRACT_PATH_TEXT(
                SYSTEM$GET_ICEBERG_TABLE_INFORMATION('{table_name}'),
                'metadataLocation'
            ) AS metadata_file_path
    """

    cursor.execute(query)

Snowflakeからのデータ抽出処理においてAWSとGoogle Cloudにおいて実行するタスクに大きな違いはありませんが、プラットフォームのセットアップ観点ではややGoogle Cloudの方がシンプルで管理しやすいというメリットがあります。

具体的には、AWSでもGoogle Cloudでも鍵そのものを扱わないという点では双方同じ思想ですが、AWSではSnowflake側が提示する認証情報（STORAGE_AWS_EXTERNAL_ID）を指定したポリシーを自分たちで作成する手順が必要なのですが、Google Cloud StorageではSnowflakeが提示するサービスアカウントに対してバケットの利用権限を付与してあげるだけでよいため、シンプルにセットアップを進めることができます。

② Load処理

前述の通り、本プラットフォームは加⼯したデータを全社的に活⽤してもらうことを目指しています。そこで、データ処理の最後のステップとして今回作成したデータセットをSnowflake上に書き戻す処理を実施しています。

本処理では、「① Extract(抽出)処理」と逆のプロセスを実施しています。具体的には、BigQuery上でIceberg化したメタデータをGoogle Cloud Storage経由でSnowflakeに連携し、Snowflake上でIcebergテーブルとして読み込み可能な処理を実装しています。

このように全社共通プラットフォームのSnowflake上でも閲覧できる一方で、データの実体はGoogle Cloud Storage上のみに存在するという状態を維持することができます。プラットフォーム側とSnowflake側の両方でデータを二重に持つ必要が無くなるので、その分だけSnowflakeのストレージ利用料金を削減することができますし、プラットフォーム側とSnowflake側でデータの差分がでる可能性が無くなりますので運用負荷の軽減という面でも有効な方法だと思っています。

③ Transform処理

本プラットフォームのデータ変換処理の特徴はお客様の声データに対してLLM処理を実施している点です。

毎日数千件ほど連携されるお客様の声データひとつひとつに対してLLM処理を実施していく必要があるため、処理をスケーリングしながら安定的に効率良く処理を実施する必要がありました。

今回はBigQueryの外部関数（Remote Function）機能を用いて実現しています。具体的にはVertex AIを呼び出すための処理をPythonにてコード化したものを、Cloud Run Functionsにデプロイし、そのCloud Run Functionsを外部関数として登録することで、BigQueryからスムーズにLLM処理を呼び出しつつもスケーリングしながら安定的にデータを処理することを実現しました。

BigQueryの外部関数（Remote Function）からCloud Run Functionsを並列で呼び出し、さらにCloud Run Functions内にデプロイされているコードでVertex AIのAPIを並列で呼び出している。

このプラットフォーム上では1つの声データに対して2つのLLM処理を行っています。

ラベリング処理：声テキスト本文とプロンプトをGoogle Gemini（モデル：gemini-2.5-flash）に渡し、そのテキストの内容が「称賛」「問題指摘」「要望」など6種類のどのカテゴリに属するかを判定してフラグ付け
ベクトル化処理：声テキストの本文をGoogle Gemini（モデル：gemini-embedding-001）に渡し、そのテキストをベクトル変換

この記事の執筆時点では、AWS Redshiftと比較してBigQueryではSQLだけでモデルを呼び出すことに力を入れているように感じています。ただ、現時点ではデータセットとVertex AIのロケーションが異なる場合は直接モデルを呼び出すことはできないため、今後のアップデートでその部分が改善されることを期待しています。

苦労した点、工夫した点

1. LLMの呼び出し部分のチューニングに苦労した

実装当初、LLM処理の部分で思っていた以上に処理速度が出ないことがあり、その原因を調査したり適切なリソース設定をする点に苦労しました。

調査の結果、デフォルトで有効となっていたThinking processオプションを無効化することで、LLMの処理速度が大幅に改善しました。 Thinking processは複雑な課題解決に有効な機能ですが、本ユースケースでは比較的単純なラベリング処理を速度重視で実行する必要があったため無効化しています。*2

PythonによるGemini API呼び出しの実装例

    resp = await aio.models.generate_content(
        model=MODEL_ID,
        contents=user_prompt,
        config=GenerateContentConfig(
            thinking_config=types.ThinkingConfig(thinking_budget=0), #この部分を0に指定
            response_mime_type="text/plain",
            temperature=0,
            seed=42,
        ),
    )

しかし、この改善により処理速度が想定より高速化した結果、Vertex AIの分間レートリミットに抵触する事態が発生しました。

原因はベクトル化処理の負荷が想定以上に高いことによりCPU利用率の増加しているためでした。 Cloud Run FunctionsはCPU利用率をトリガーとしてスケーリングすることを踏まえ、適切なコンテナサイズを指定し、Vertex AIのレートリミットに抵触しない範囲の処理量に調整して解決を図りました。

2. Icebergのデータ削除の方法を工夫した

Icebergテーブルの特性上、何らかのレコードを直接Icebergテーブルから削除しても、メタデータ上ではそのデータはまだ存在しているため、次の処理を実⾏するタイミングでそのメタデータを参照してしまい削除されたデータが復元されます。

Icebergテーブル上からデータを削除したい場合はメタデータ自体にもデータが削除されたという事を伝えますが、この場合メタデータ上にもこのデータが削除されたという情報が過去の履歴が残ります。

上記の内容を読むと一見通常の操作に見えますが、裏を返すとメタデータ上の情報も含めてデータを完全に削除するには通常のdeleteやdropのような操作では不十分であるということになります。

例えば、通常は入ってはいけないような機密度の高い情報が紛れ込んでしまい、そのデータを完全に削除する必要がある場合は物理的に削除する必要があります。そこで本プロジェクトでは上記のような場合に備えて、そのデータを削除するためのマニュアルを作成しました。

手順の詳細はこの場では割愛いたしますが「Google Cloud Storage上の対象となるメタデータを直接削除し、上流のテーブルから順に再作成していく」というのが基本の進め方になっています。

メタデータをバケット上から直接削除する部分については、Spark REST APIを組み込むなどの追加対応を行うことでより楽に対応することができるのですが、そのためだけにSparkを導入するのもコスト面などでハードルがあるためバランスとして難しい部分だと感じています。

今後の展望

取り込みデータの拡張と可用性の向上

今後はお客様の声データだけでなく、X（旧Twitter）上のドコモに関する口コミ情報や、ドコモショップなどの実店舗での応対履歴などの情報もこのプラットフォームに取り込み、現在よりもさらに幅広い分析ができるようにアップデートを進めていく予定になっています。

今後は処理するデータ量が増えていく中で早く確実にデータを処理することが求められており、それに対応できるようなサービス構成を目指しています。例えば、確実にデータを処理していくことに対しては、直近のGoogle Cloud Next'25 で発表された「Vertex AI Global Endpoint」の活用を検討しています。

従来はリージョンごとのエンドポイントを指定してVertex AIを利用していましたが、この「Vertex AI Global Endpoint」を利用することで、リージョン障害が発生したりGoogle Cloud上でリソース枯渇が発生したりした際には利用可能な別リージョンにルーティングされるため、可用性を高めることができそうだと考えています。