Introducing Attribute™ — trace every AI token to its owner
Cloud Intelligence™Cloud Intelligence™

Cloud Intelligence™

データベースをIcebergデータレイクへ移行する（第2回）

By Sayle MatthewsJul 13, 202616 min read

このページはEnglish、Deutsch、Español、Français、Italiano、Portuguêsでもご覧いただけます。

シリーズについて

本記事は、データベースをIcebergベースのデータレイクへレプリケートするシリーズの続編です。第1回では、その理論・方法・目的を解説しました。第2回となる本稿では、コード、コマンド、そして本番投入で得られた学びを交えながら、実際の実装方法を紹介します。

実装の前提条件

必要な前提条件は、Kubernetesクラスタを立ち上げられるクラウド環境（またはオンプレミス環境）、Debezium Serverに対応したソース、そしてIcebergテーブルのデータを格納するGCSやS3などのBLOBストアがあることだけです。本記事ではソース以外のセットアップ手順をGCP上で説明します。

この基本例を動かすだけであれば、Kubernetesの深い知識はほとんど不要です。実行すべきコマンドはすべて掲載しています。もちろん本番規模で運用する場合はこの基本例を超える知識が求められますが、この例ではKubestronaut（Kubernetes熟練者）である必要はありません。

なお、Kubernetes以外の方法でもデプロイ可能です。Kubernetesクラスタ上でこのアプリケーションだけを単独で運用する場合は、コスト面からKubernetes以外の選択肢も検討する価値があります。フルのKubernetesクラスタが不要で本アプリケーションだけを動かすのであれば、GCEやEC2などのコンピュートプラットフォーム上でスタンドアロンコンテナとして動かすことを検討してみてください。

実装プラン

本実装は以下の手順で進めます:

  1. ソース（今回はPostgres）を準備する
  2. バケット／出力先の権限を設定する
  3. Kubernetesデプロイメントを自分の環境に合わせて修正する
  4. Kubernetesデプロイメントをデプロイする
  5. 動作確認を行う

プランはあえてシンプルに設計しています。今回はあくまで概念実証（PoC）を想定した基本ユースケースですが、本番運用に向けて改善できるポイントは随所で指摘していきます。

私はよく「The BigQuery Dude」と呼ばれているので、慣れ親しんだGCP上で実装していきます。ただし、この例はAWSやAzureでも簡単にデプロイできます。今回はDebezium ServerをGKE上にデプロイし、ソースにはCloud SQL上のPostgres、格納先にはGCS上のIcebergテーブルを使いますが、これらはいずれもAWSとAzureに同等のサービスがあります。

コードは可能な限りプラットフォーム非依存で書いていますが、一部はGCP固有のものになります。以下の「gcloud」コマンドと「gs://」プレフィックス以外は、他のプロバイダー上のあらゆるKubernetes、ストレージ、Postgresへそのまま移植できます。

サンプルデータには、私の世代の多くがSQLを学ぶ際に使った有名なNorthwindデータベースを使います。コードはGitHubのこちらから、SQLファイルに直接アクセスしたい場合はこちらからダウンロードできます。データはpublicスキーマにあり、検索・置換のしやすさとそのままテスト実行できる利便性から、すべて「my_user」というユーザーで実行しています。

実装（ソースデータベースのレプリケーション）

実装の最初のステップは、データベースの準備です。この手順はこれまでにも数多く解説されているため、詳細は割愛します。今回のテストケースでは、GCPのCloud SQL上でPostgreSQLを使う想定です。PostgreSQL以外を使う場合は、こちらで対応データベース向けの手順を確認してください。

まず、wal_levelフラグをlogicalに設定する必要があります。これは主にセルフホストやCloud SQL以外のインスタンスで必要な作業です。Cloud SQLではマネージドレイヤーで制御されており、「logical_decoding」フラグと呼ばれます。設定するにはインスタンスを編集し、フラグ内でcloudsql.logical_decodingがOnになっていることを確認してください。まだ追加されていない場合は追加し、値をOnにします。

次はレプリケーション用ユーザーの設定です。監視しやすくするためにも、わかりやすい名前で専用ユーザーを新規作成することを強くおすすめします。よく見かける名前はdebeziumやcdcなどですが、お好みでどうぞ。

新規ユーザーを作成するには、以下のコマンドを実行します（エラーが出た場合は下の説明を参照してください）:

CREATE USER my_user WITH REPLICATION LOGIN PASSWORD '<password>';

上記コマンド実行時に、ユーザーにREPLICATIONロールを付与できないというエラーが出た場合は、以下のコマンドを実行してください（権限が必要です。権限がない場合は管理者に依頼するか、postgresユーザーを使ってください）:

ALTER USER my_user WITH REPLICATION;

ここからは、どのテーブルをレプリケートするかを決める大事な段階です。ここではシンプルさのために単一テーブルを対象にします。以下のコマンドで対象テーブルとスキーマを指定し、選択したユーザーにアクセス権を付与します。カンマ区切りで必要なだけ指定できます:

-- 「my_publication」の部分はお好みの名前に変更してください
CREATE PUBLICATION my_publication
FOR TABLE public.orders;


-- 上で選択した各スキーマに対して以下を実行
GRANT SELECT ON ALL TABLES IN SCHEMA public TO my_user;
GRANT USAGE ON SCHEMA public TO my_user;
ALTER DEFAULT PRIVILEGES IN SCHEMA public
    GRANT SELECT ON TABLES TO my_user;

本記事で扱う範囲を超えた、より実践的な用途向けに、スキーマ内の複数テーブルまたは全テーブルを対象にするSQL構文は以下の通りです:

-- 複数テーブルの場合
CREATE PUBLICATION my_publication
FOR TABLE public.orders, public.order_details;


-- スキーマ内の全テーブルの場合
CREATE PUBLICATION my_publication
FOR TABLES IN SCHEMA public;


-- 上で選択した各スキーマに対して以下を実行
GRANT SELECT ON ALL TABLES IN SCHEMA public TO my_user;
GRANT USAGE ON SCHEMA public TO my_user;
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO my_user;

最後に、レプリケーションスロットを作成します。レプリケーションスロットとは、変更されたデータを常時レシーバーへパブリッシュし続けるプロセスのことです。あとで実行や検索・置換をしやすくするため、ここではmy_replication_slotという名前にしています。以下のコマンドを実行してください:

SELECT PG_CREATE_LOGICAL_REPLICATION_SLOT('my_replication_slot', 'pgoutput');

この時点で、ソースデータベース側のレプリケーション設定は完了です（本記事に沿ってPostgresを使っている前提です）。

重要な注意点: このレプリケーションスロットは、実際に使用していなくても常時稼働し、変更データをディスクに蓄積し続けます。使わないのであれば必ず削除してください。放置すると 確実に ディスク容量を消費し続け、データベースインスタンスのコストが膨らんでいきます。

サービスアカウントのセキュリティに関する注意

本記事では、GKEクラスタおよびDebezium ServerサービスをデフォルトのCompute Engineサービスアカウントで動かしています。これはあくまでデモンストレーション用であり、最小権限の原則にまで踏み込まないための措置です。本番環境では 絶対に デフォルトサービスアカウントで運用しないでください。必要な権限だけを付与した新規サービスアカウントを作成してください。この詳細は本記事の範囲を超えるため、まずはGoogleドキュメントの該当記事から掘り下げていくことをおすすめします。

また、デモンストレーションの都合上、セキュリティよりも簡潔さを優先している箇所はその都度明記しています。将来のセキュリティ問題を回避するためにも、こうした点を把握し、より安全な方法を採用してください。お願いします！

実装（環境変数の準備）

このステップは、皆さんの作業をラクにするために追加しました。私自身、最初はこの手順を挟んでいませんでしたが、公開前の再検証時に「これはあった方が絶対にラクだ」と気づいて追加した次第です。目的は、以下のコマンドをステップごとに値を書き換えずに実行できるようにすることです。破線より下は生成される値なので、変更しないでください。

なお、ここには認証情報が含まれます。本番環境で行う場合はSecret Managerなどより安全な方法を使ってください。

最後のコマンドセットでは、確認のためパスワード以外の各値をエコー出力します。エラーが出た場合は、値が正しく設定されているか確認してください。

以下がコマンドです。各項目の説明はコメントに記載しています:

Terminal window
# プロジェクト名
PROJECT_NAME=<project_name>
# Icebergテーブルを格納するバケット名
BUCKET_NAME=<bucket-name>
# 上記バケット内でIcebergテーブルを格納するベースパス
# 通常はテーブル名やスキーマ名を指定します。
# 不明な場合は空にしておくと、バケットのルートに書き込まれます。
BUCKET_PATH=<bucket_path>
# クラスタ名。上のコマンドではcdc_clusterを使用しています
CLUSTER_NAME=<cluster_name>
# クラスタリージョン。上のコマンドではus-central1を使用しています。
# Cloud SQLインスタンスと同じリージョンを指定してください
CLUSTER_REGION=<cluster_region>
# Cloud SQLインスタンス名
SQL_INSTANCE_NAME=<cloud_sql_instance_name>
# 上で作成したデータベースユーザー名
DATABASE_USER=<database_user>
# 上で作成したユーザーのパスワード
DATABASE_PASSWORD=<database_password>
# 上のコマンドを適用したデータベース名
DATABASE_NAME=<database_name>


#----------------------------------------------------------------------
# この線より下は変更しないでください。
# 追加の値を取得し、後で利用する生成環境変数を設定しています。
gcloud config set project $PROJECT_NAME
PROJECT_NAME=$(gcloud config get-value project)
PROJECT_ID=$(gcloud config get-value project 2>/dev/null)
SQL_NAME=gcloud sql instances describe $SQL_INSTANCE_NAME --format='value(connectionName)'


# 以下で上記の値が正しく設定されているかを確認するためエコー出力します
echo "Project Name: $PROJECT_NAME"
echo "Project ID (Generated from gcloud): $PROJECT_ID"
echo "Bucket Name: $BUCKET_NAME"
echo "Bucket Path: $BUCKET_PATH"
echo "Full Bucket Path: gs://$BUCKET_NAME/$BUCKET_PATH"
echo "Cluster Name: $CLUSTER_NAME"
echo "Cluster Region: $CLUSTER_REGION"
echo "SQL Instance Name (Given): $SQL_INSTANCE_NAME"
echo "SQL Instance Name (Received from gcloud): $SQL_NAME"
echo "Database Username: $DATABASE_USER"
echo "Database Password (hidden): ******"
echo "Database Name: $DATABASE_NAME"

実装（バケット権限）

GCSバケットに読み書きするには、まずバケットを作成する必要があります。そして、次のセクションで実装するGKEクラスタを動かすサービスアカウントに、いくつかの権限を付与する必要があります。

IAM要件をGoogleの推奨より簡素にできないかを検証しましたが、テストの結果、このサービスアカウントに付与するには以下の2つの事前定義ロールを組み合わせるのがベストという結論に至りました:

GoogleはBigLakeの実装に関してもこの点に触れており、BigLakeを使うのであればそのページを一読することをおすすめします。ただし、そこで省かれているのは、Icebergテーブルを書き込むためにStorage Object Adminロールの権限が必要になるという点です。とにかくこのロールを付与し、権限の組み合わせで試行錯誤する沼にはまらないでください。信じてください。

以下のコマンドを実行して、バケットを作成し、サービスアカウント（ここではデフォルトのCompute Engine用）にIAMポリシーバインディングを適用します。必要に応じて変更してください:

Terminal window
gcloud storage buckets create gs://$BUCKET_NAME
gcloud storage buckets add-iam-policy-binding gs://$BUCKET_NAME \
    --member="serviceAccount:$PROJECT_NAME[email protected]" \
    --role="roles/storage.objectAdmin"
gcloud storage buckets add-iam-policy-binding gs://$BUCKET_NAME \
    --member="serviceAccount:$PROJECT_NAME[email protected]" \
    --role="roles/storage.legacyBucketReader"

コマンドの内容はシンプルです。最初のコマンドはgcloudコマンドを正しいプロジェクトに設定するだけ、2つ目でバケットを作成、そして残りの2つでプロジェクト内のデフォルトのCompute Engineサービスアカウントに必要な権限を付与します。

実装（Kubernetesクラスタの作成）

この実装では、Google Kubernetes Engine（GKE）Autopilotを使ってKubernetesクラスタを作成し、HelmチャートでDebezium Serverを載せていきます。GKEに慣れている方や、すでに別のクラスタを稼働させている方は勝手がわかっていると思いますので、以下のロード用コマンドまで読み飛ばしていただいて構いません。

まず、デフォルトのCompute Engineサービスアカウントに必要なロールを1つ追加します。コマンドは以下の通りです:

Terminal window
gcloud projects add-iam-policy-binding $PROJECT_ID \
    --member="serviceAccount:$PROJECT_NAME[email protected]" \
    --role="roles/container.defaultNodeServiceAccount"

次にGKE Autopilotクラスタ本体を作成します。以下のコマンドを実行するだけです。

Terminal window
gcloud container clusters create-auto $CLUSTER_NAME \
    --location=$CLUSTER_REGION \
    --project=$PROJECT_ID

起動には少し時間がかかりますが、数分ほどで完了通知が出て、GCP ConsoleのGKEセクションでクラスタを確認できるようになります。

実運用時のネットワーク構成に関する注意

実際の運用では、Cloud SQLを別リージョンで動かしていたり、プロジェクトをまたいだ通信を扱ったりする場面が多くあります。私がまずおすすめしたいのは、GCPのネットワーク固有の「クセ」を回避するために、プライベートIPとCloud SQL Proxyを可能な限り活用することです。

実装（Kubernetesクラスタへのデプロイ）

いよいよ楽しいパート、GKE上へのDebezium Operatorのデプロイです。以下のコマンドを実行し、ダウンロードと起動が終わるまで数分待つだけです:

Terminal window
# gcloudをGKEクラスタで認証。これで以降kubectlが実行できるようになります
gcloud container clusters get-credentials $CLUSTER_NAME --region $CLUSTER_REGION


# Helmリポジトリを追加し、ローカルのHelmソースリストを更新
helm repo add debezium https://charts.debezium.io
helm repo update


# 専用のnamespace（例: debezium-system）にoperatorをインストール
# namespaceが存在しなければ作成されます
helm install debezium-operator debezium/debezium-operator \
  --namespace debezium-system \
  --create-namespace

注意: 各コマンドが何をしているかを示すため、コメントを残しています。挙動の理解や潜在的な問題の切り分けに役立ててください。

デプロイの成否を確認したり、最悪の場合はデプロイ失敗の原因を切り分けたりするには、以下のコマンドで作成したnamespace内のpod一覧と関連ログを表示できます:

Terminal window
# debezium-system namespace内のpodを取得
kubectl get debeziumserver -n debezium-system
# podのログを表示
kubectl logs -l app.kubernetes.io/name=debezium-server -n debezium-system

ここまで来たら、いよいよ本番のデプロイです！

以下のコマンドは少し長いですが、やっていることは比較的シンプルで、Debezium Serverとその設定をCloud SQL Proxy付きでデプロイしているだけです。Kubernetesに詳しい方向けに言えば、Debezium Serverを実行するpodにCloud SQL Proxyをサイドカーとしてデプロイしています。

準備ができたら、次のコマンドを実行してください:

Terminal window
kubectl apply -f - <<EOF
apiVersion: debezium.io/v1alpha1
kind: DebeziumServer
metadata:
  name: debezium-sql-proxy
  namespace: debezium-system
spec:
  image: debezium/server:latest
  # --- Sidecar Configuration ---
  runtime:
    containers:
      - name: cloud-sql-proxy
        image: gcr.io/cloud-sql-connectors/cloud-sql-proxy:latest
        args:
          - "--port=5432"
          - "$PROJECT_NAME:$CLUSTER_REGION:$SQL_NAME"
        securityContext:
          runAsNonRoot: true
  # --- Connector Configuration ---
  source:
    connector.class: io.debezium.connector.postgresql.PostgresConnector
    config:
      # プロキシが同一Pod内にあるためlocalhostに接続
      database.hostname: "127.0.0.1"
      database.port: "5432"
      database.user: "$DATABASE_USER"
      database.password: "$DATABASE_PASSWORD"
      database.dbname: "$DATABASE_NAME"
      topic.prefix: "cdc"
      # Cloud SQL Postgresでは必須
      plugin.name: "pgoutput"
  # --- Iceberg Sink Configuration ---
  sink:
    type: iceberg
    config:
      # Icebergカタログの種類（GCSのみの構成にはHadoopが最適）
      debezium.sink.iceberg.catalog.type: hadoop
      debezium.sink.iceberg.catalog.warehouse: gs://$BUCKET_NAME/$BUCKET_PATH


      # GCS FileSystemの設定
      debezium.sink.iceberg.fs.defaultFS: gs://$BUCKET_NAME
      debezium.sink.iceberg.io-impl: org.apache.iceberg.gcp.gcs.GCSFileIO


      # テーブルの挙動: upsert（変更をマージ）またはappend
      debezium.sink.iceberg.upsert: "true"
      debezium.sink.iceberg.upsert.keep-deletes: "false"


      # コミット間隔（新しいIcebergスナップショットを作成する頻度）
      debezium.sink.iceberg.table.commit.interval-ms: "60000"
EOF

出力の確認（お待ちかね！）

いよいよ動作確認です！Postgresに接続し、先ほどコマンドを適用したテーブルに行をINSERTしてみてください。Postgres側で必要な作業はこれだけで、以降のレプリケーションが自動でトリガーされます。GCS上のIcebergテーブルへ、データが直接書き込まれ始めるはずです。

確認にはBigQueryを使ってIcebergテーブルにクエリを実行します。BigQuery上でBigLake外部テーブルを作成する形になります。BigQueryに馴染みのない方向けに言えば、これは「GCSバケット上に特定フォーマットのテーブルがあるので、GCS読み取り用のBigLakeという抽象化レイヤーを介してBigQueryからクエリしたい」とBigQueryに伝えるようなイメージです。

手順としては、GCP Consoleを開いてBigQueryセクションに移動します。BigQuery Studioと呼ばれるテキストエディタウィンドウが開きます。

そこに以下のコマンドを貼り付け、バケットのリージョン、バケット名、上で設定したバケットパスに合わせて修正します:

Terminal window
CREATE SCHEMA IF NOT EXISTS `testing_biglake_dataset`
OPTIONS(
  location = '<bucket_region>'
);


CREATE OR REPLACE EXTERNAL TABLE `testing_biglake_dataset.my_testing_iceberg_table`
WITH CONNECTION `testing_biglake_dataset.my_connection`
OPTIONS (
  format = 'ICEBERG',
  uris = ['gs://<bucket_name>/<bucket_path>/metadata.json']
);

これを実行すると、専用のデータセットが新規作成されます。こうしておけば、あとで一括削除して後片付けもラクになります。

クエリが完了したら、BigQuery Studioエディタペインの上部にある「新しいクエリ」タブをクリックし、以下のSQLを貼り付けて実行してください:

SELECT * FROM testing_biglake_dataset.my_testing_iceberg_table TABLESAMPLE SYSTEM (15 PERCENT)

これで、挿入したデータが数行返ってくるはずです。15%のテーブルサンプリングを行っているため、全データは返しません。BigQueryは高額な請求を発生させる能力にも定評があるので、非常に小さなクエリで済むよう予防策を取っています。

この時点で、PostgresからIcebergへのデータレプリケーションの実装は無事完了です！次のステップは、これを複数のテーブル（あるいはスキーマ）に対して行い、すべてのデータを「データベース非依存」なデータストアに集めていくことです。

まとめ

これは単一テーブルを単一のIcebergテーブルに書き込むだけの基本的な例ですが、その威力は絶大です。膨大なトランザクションが行き交う販売データベースまるごとを、リアルタイムでデータウェアハウスに書き込み、最新の売上データをダッシュボードやレポートで即座に確認できる――そんな世界を想像してみてください。

何より素晴らしいのは、クエリするデータウェアハウスが何でも構わないという点です。BigQuery、ClickHouse、Snowflake、Databricksなど、あるいはそれらの組み合わせでも問題ありません。プラットフォーム非依存の構成を保つことは、技術面でもコスト面でも競争力を維持するために、今すぐ取り組める最良のプラクティスの1つです。

これはDoiT Internationalが日々扱い、アドバイスしているテーマのほんの一例です。もっと詳しく知りたい方や、クラウドコスト・クラウド運用の面でどのようにお手伝いできるかを知りたい方は、ぜひお気軽にお問い合わせください。

クリーンアップ（必要に応じて）

最後に、テスト後にコストを抑え、本番運用の検討に移るために構成を削除したい場合は、以下のコマンドでGKEクラスタを「まるごと消し去る」ことができます:

Terminal window
gcloud container clusters delete $CLUSTER_NAME \
    --location=$CLUSTER_REGION \
    --project=$PROJECT_ID

作成したBigQueryデータセットを削除するには、BigQuery Studioで以下を実行します:

DROP SCHEMA IF EXISTS `testing_biglake_dataset`;

すべて削除する場合、この手順は 絶対に 忘れないでください。Postgresインスタンスからレプリケーションスロットを削除する作業です。これを行わないと、Postgresのストレージコストが時間とともに増加してしまいます。以下のコマンドはPostgres内で実行する必要があるのでご注意ください:

SELECT pg_drop_replication_slot('my_replication_slot');

How We DoiT

DoiT Internationalでは、こうした課題に日々取り組んでおり、データプロジェクトを進めながらコストを抑える方法についてのご相談も絶えずいただいています。

プロジェクトを最も効果的かつコスト効率の高い形で実装するお手伝いをすることは、私たちがお客様に対して掲げるミッションの一部です。こうしたケースへの対応から、最良のFinOpsソリューションの提供まで幅広く手掛けており、少々身びいきかもしれませんが、私たちはこれを非常に得意としています。

さらに、コスト最適化のニーズに応えるベストインクラスのツールも提供しています。Kubernetes向けのPerfectScale、そしてBigQuery・Databricks・Snowflake向けのSelectなどです。