最終更新日 2025年03月10日(月)

この記事では、Heroku Postgres イベント用に Change Data Capture (CDC) を設定し、Private Space または Shield Private Space にプロビジョニングされた Apache Kafka on Heroku アドオンにそれらのイベントをストリーミングする方法について説明します。このプロセスは、大きく分けて 3 つの手順で構成されます。

1. Private Space または Shield Private Space にアプリを作成する
2. Private または Shield の Heroku Postgres アドオンと、Private または Shield の Apache Kafka on Heroku アドオンを新しいアプリにプロビジョニングする
3. ストリーミングデータコネクターを作成して Postgres から Kafka への CDC イベントを有効にする

ストリーミングデータコネクターを最適に設定する方法についての詳細は、「Heroku のストリーミングデータコネクターのベストプラクティス​」を参照してください。

Heroku アプリの設定

まず、Private Space または Shield Private Space を作成します。Space が利用可能になったら、その Space にアプリを作成できます。

$ heroku spaces:create --region virginia --team my-team-name --space myspace
$ heroku spaces:wait --space myspace
$ heroku apps:create --space myspace my-cdc-app

Heroku アドオンの設定

次に、2 つの Private または Shield のデータアドオンをアプリにアタッチする必要があります。

$ heroku addons:create heroku-postgresql:private-7 --as DATABASE --app my-cdc-app
$ heroku addons:create heroku-kafka:private-extended-2 --as KAFKA --app my-cdc-app

アドオンのプロビジョニングの進行状況は、次のようにして監視できます。

$ heroku addons:wait --app my-cdc-app

アドオンが利用可能になったら、スキーマとデータを Postgres データベースにインポートします。

Heroku のストリーミングデータコネクターの設定

Private Space または Shield Private Space アプリで Heroku Postgres アドオンと Apache Kafka on Heroku アドオンを設定したら、コネクターをプロビジョニングできます。

最初に、CLI プラグインをインストールします。

$ heroku plugins:install data

コネクターを作成​するには、いくつかの情報を収集する必要があります。

1. Kafka アドオンの名前
2. Postgres アドオンの名前
3. そこで発生したイベントをキャプチャする Postgres テーブルの名前
4. (オプション) キャプチャイベントから除外​する列の名前

Postgres データベースでイベントをキャプチャするためには、いくつかの要件が満たされている必要があります。

• データベースのエンコードが UTF-8 であること
• その時点でテーブルが存在していること
• テーブルにプライマリキーがあること
• テーブルがパーティション分割されていない​こと
• テーブル名に [a-z,A-Z,0–9,\_]​ 以外の文字が含まれていないこと
• Kafka Formation で直接の Zookeeper アクセスが無効になっていること

次に、コネクターを作成できます。Postgres アドオンと Kafka アドオンの名前に加えて、データベースキャプチャイベントに含める完全修飾テーブルのリストが必要です。

$ heroku data:connectors:create \
    --source postgresql-neato-98765 \
    --store kafka-lovely-12345 \
    --table public.posts --table public.users

プロビジョニングの完了まで約 15 ～ 20 分かかることがあります。コネクターのプロビジョニングの進行状況は、次のようにして監視​できます。

$ heroku data:connectors:wait gentle-connector-1234

ストリーミングデータコネクターでは、データの変更をキャプチャするために選択したテーブルごとに、Kafka クラスターにトピックが作成されます。コネクターが利用可能になったら、新しく作成された Kafka トピックなどの詳細を表示​できます。

$ heroku data:connectors:info gentle-connector-1234
=== Data Connector status for gentle_connector_1234
Name:   gentle_connector_1234
Status: available

=== Configuration
Table Name   Topic Name
public.posts gentle_connector_1234.public.posts
public.users gentle_connector_1234.public.users

これらのトピックは次のパラメータ​で設定されます。

• partition_count​: 32
• replication_factor​: 3
• cleanup_policy​: delete
• retention_time_ms​: 24 時間

プリンシパル (Kafka ユーザー) は、ストリーミングデータコネクターによって作成されたテーブルトピックに対して、Read​ および Describe​ アクセス権​を持ちます。Write​、Delete​、および Alter​ 操作は拒否されます。

この機能によって、各コネクターのハートビートトピック​も作成されます。プリンシパルは、ハートビートトピックに対しても Read​ および Describe​ アクセス権を持ちます。

コネクターの管理

作成したコネクターを管理するために、いくつかの操作を実行できます。

一時停止または再開

新しいイベントの処理を一時停止​できます。コネクターを一時停止すると、再開​するまでの間、追加のレコードのポーリングが停止されます。2 つの状態は、次のようにして簡単に切り替えることができます。

# to pause
$ heroku data:connectors:pause gentle-connector-1234

# to resume
$ heroku data:connectors:resume gentle-connector-1234

設定の更新

コネクターに関連付けられた特定のプロパティを CLI から変更できます。これらのプロパティには以下の値が含まれます。

たとえば、tombstones.on.delete​ を false​ に更新できます。

$ heroku data:connectors:update gentle-connector-1234 \
  --setting tombstones.on.delete=false

Heroku によって管理される設定

ほとんどの設定プロパティは Heroku によって完全に管理され、必要に応じて変更されます。

テーブルおよび除外される列を更新する

除外される列のほかに、コネクターの Postgres テーブルも変更できます。

たとえば、public.parcels​ テーブルを追加し、public.posts​ テーブルを削除できます。

$ heroku data:connectors:update gentle-connector-1234 \
  --add-table public.parcels \
  --remove-table public.posts

新しいテーブルは、設定に概略を示すものと同じ要件に従う必要があります。

同様に、除外される列を追加および削除できます。

$ heroku data:connectors:update gentle-connector-1234 \
  --exclude-column public.parcels.address \
  --remove-excluded-column public.posts.keys

Heroku Postgres データベースのバージョンのアップグレード

ストリーミングデータコネクターを使用して Heroku Postgres データベースをアップグレードするには、新しくアップグレードしたデータベースでコネクターを再作成する必要があります。コネクターがイベントを新しいデータベースにストリーミングするように、システムを再構成するための追加手順があります。アップグレード自体はフォロワーデータベースと pg:upgrade​ を使用して行われます。これらの手順は、ストリーミングデータコネクターがアップグレードプロセスにどのように適合するかを示します。

ストリーミングデータコネクターを使用して、データベースをアップグレードする手順は次のとおりです。

1. 現在のコネクターとその設定に関する情報を取得する
2. アップグレードしてプロモートする新しいフォロワーデータベースを準備する
3. 古いデータベースへの書き込みを無効にし、コネクターを一時停止する
4. アップグレードを実行する
5. 新しくアップグレードしたデータベースをプロモートまたはアタッチする
6. 新しくアップグレードしたデータベースでコネクターを再作成する
7. メンテナンスモードを終了する
8. 古いプライマリデータベースをプロビジョニング解除する

1. コネクターに関する情報を取得する

コネクターのリストと各コネクターの設定の詳細を取得する必要があります。この情報は、アップグレードしたデータベースへのコネクターを再作成する手順 6 で必要になります。

コネクターのリストを取得するには、次のコマンドを実行します。

$ heroku data:connectors --app example-app

=== Data Connector info for example-app
Connector Name: inventive-connector-83577
Kafka Add-On: example-app-kafka
Postgres Add-On: example-app-postgres
Tables: public.posts
public.comments

特定のコネクターの情報と構成の詳細を取得するには、コネクター名を指定してコマンドを実行します。

$ heroku data:connectors:info inventive-connector-83577

=== Data Connector status for inventive-connector-83577
Lag: 15 MB
Service Name: 10d5e5cb-0343-4166-80d8-f03fcc4d1e21
Status: available

=== Configuration
Table Name Topic Name
public.posts inventive_connector_83577.public.posts
public.comments inventive_connector_83577.public.comments

Your Data Connector is now available.

2. フォロワーデータベースをプロビジョニングする

次に、フォロワーデータベースを作成​し、リーダーデータベースにほぼ追いつくまで待機します。フォロワーを作成すると、アップグレードに必要なダウンタイムが最小化されます。アップグレードの 24 時間前までにフォロワーを作成することをお勧めします。

3. メンテナンスモードにしてコネクターを一時停止する

次に、アプリをメンテナンスモード​にしてアプリの dyno をスケールダウンし、アップグレード中にデータベースへの書き込みが行われないようにします。

$ heroku maintenance:on --app example-app
$ heroku ps:scale consumer=0 generator=0 web=0 --app example-app

その後、データコネクターを一時停止します。

$ heroku data:connectors:pause inventive-connector-83577

Pausing Data Connector inventive-connector-83577... done

4. フォロワーデータベースをアップグレードする

アップグレードする前に、フォロワーが pg:info​ に追いついていることを確認します。フォロワーデータベースが同期されていると、Behind By​ フィールドに 0 commits​ が表示されます。

次に、pg:upgrade​ を使用してフォロワーデータベースをアップグレード​します。このコマンドによってフォロワーでリーダーのフォローが解除され、Postgres バージョンのアップグレードが実行されます。pg:wait​ を使用して、アップグレードの進捗状況を監視できます。

5. 新しいデータベースをプロモートまたはアタッチする

DATABASE_URL​ が以前のプライマリデータベースの環境設定​だった場合は、pg:promote​ を使用して新しくアップグレードしたデータベースを新しい DATABASE_URL​ としてプロモートします。アップグレードするデータベースにデフォルトの DATABASE_URL​ とは異なる環境設定がある場合は、heroku addons:attach​ を使用して、必要なエイリアスまたはアタッチメント名で新しくアップグレードしたデータベースをプロモートします。

6. コネクターを交換する

次に、古いデータベースでコネクターを破棄し、新しくアップグレードしたデータベース用にコネクターを再作成する必要があります。破棄する前に、ステップ 1 の情報があることを確認してください。通常、コネクターの破壊を完了するには約 10 分かかります。

$ heroku data:connectors:destroy inventive-connector-83577

To proceed, type inventive-connector-83577 or re-run this command with --confirm inventive-connector-83577: inventive-connector-83577
Data Connector inventive-connector-83577 deleted successfully.
Note: We do not delete your Kafka topics automatically, because they could still contain messages which you haven't consumed. Please delete the topics manually. See heroku kafka:topics:destroy --help
Destroying Data Connector... done

コネクターが破棄されたら、新しいデータベースでコネクターを再作成​します。

7. メンテナンスモードを終了する

通常のアプリケーション操作を再開するには、Web 以外のすべての dyno をその元のレベルにスケーリングし、メンテナンスモードをオフにします。

$ heroku ps:scale consumer=1:Private-M generator=1:Private-M web=1:Private-M --app example-app

Scaling dynos... done, now running consumer at 1:Private-M, web at 1:Private-M, generator at 1:Private-M, console at 0:Private-M, rake at 0:Private-M

$ heroku maintenance:off --app example-app

アップグレード後にアプリケーションが正常に動作し、イベントが新しい接続からストリーミングされていることを確認することをお勧めします。kafka:tail​ を実行して、コネクタートピックに最新メッセージを表示します。

8. 古いプライマリデータベースをプロビジョニング解除する

データベースをアップグレードした後、古いプライマリデータベースを必ずプロビジョニング解除してください。

$ heroku addons:destroy HEROKU_POSTGRESQL_LAVENDER --app example-app

コネクターの破棄

コネクターの破棄​を CLI から実行できます。

$ heroku data:connectors:destroy gentle-connector-1234