IBM TechXchange Integration Group Japan

View Only

Back to Blog List

IBM Data ReplicationのChange Data Captureを使って、Db2のデータをIBM Event Streamsに連携する（設定編）

By HIDEO SAITOH posted Fri September 05, 2025 02:51 AM

IBM Data ReplicationのChange Data Captureを使って、Db2のデータをIBM Event Streamsに連携する（設定編）

この記事では導入編に続き、設定の手順を書いていきます。

ソース（キャプチャーする）のDb2側の設定

Management Consoleから、新規にDatastoreを作成していきます。
Access Managerタブを選択し、Add new datastoreをクリックします。
必要な項目を入力します。
pingボタンをクリックすると、プロパティが自動で入力されます。
Connecction Parametersをクリックして、Db2のユーザー情報を入力します。
登録されると、以下のようにデータストアにリストされます。
データストアに対して、ユーザーを登録します。
Assign User/Group...を選択します。
Adminがでてくるので、これを選択します。
Db2にアクセスする情報が入っています。

ターゲットのEvent Streams(Kafka)側の設定

Management Consoleから、新規にDatastoreを作成していきます。
Access Managerタブを選択し、Add new datastoreをクリックします。
Datastoreの設定では、ホスト名はipv6の::1を設定し、ポートフォワードしている21701を設定して、pingをクリックして、プロパティを設定します。
Connection Parameterは、Kafka側のインスタンス作成に指定した、tsuserを指定します。
Event StreamsにもAssgin Userの設定を行います。
Adminを選択
tsuserが設定されています。

サブスクリプションの作成

サブスクリプションでは、ソースとターゲットを選択して、連携の対象を作成していきます。

Configurationタブから、Create new subscriptionを選択します。
必要な情報を入力していきます。
サブスクリプションの名前とターゲットのESを選択します。
New Projectで新たにプロジェクトを作ります。
OKボタンで作成します。
自動で、テーブルのマッピングを行うか尋ねてきます。Yesボタンで次にすすみます。
一つのテーブルを送信する設定をおこなうので、Custom Kafka mappingを選択します。
キャプチャーしたいTABLEを選択します。今回は、EMPLOYEEを選択します。
Key項目として、EMPNOを選択します。
まだ、Event Streams(Kafka)の設定ができていないので、Return to current viewを選択してFinish
サブスクリプション定義ができると、下記のようになります。

Event Streamsへの接続情報の設定

Subscriptionを選択して、Kafka propertiesを入力します。ここで設定する値は、実際には使われず。後で設定するkafkaproducer.propetiesやkafkaconsumer.propertiesが接続情報になります。
この画面では、localhost:2181など適当なものを入れることになります。

Event Streamsの接続情報の入手

Event StreamsのUIにアクセスし、このクラスターに接続をクリックします。
SCRAM資格情報を生成します。
資格情報名に、es-userと入力し、メッセージのプロデュース、メッセージのこんしゅーむ、およびトピックおよびスキーマの作成を選択します。
すべてのトピックを選択
すべてのコンシューマー・グループを選択します。
トランザクションIDなしを選択して、資格情報を生成します。
es-userのSCRAMパスワードが生成されました。あとでつかうので、記録しておきます。
次に、PKCS12の証明書をダウンロードします。
es-cert.p12というファイルがダウンロードされました。パスワードは以下のように生成されました。
これも後で利用するので、どこかに控えておきます。

IBM Event Streamsに接続するには、これを参考にkafkaproducer.propertiesとkafkaconsumer.propertiesを作成します。

外部の Kafka リスナーのブートストラップ・アドレス。
オプション: スキーマ・レジストリー・エンドポイント。
SCRAM ユーザー名とパスワード
オプション: CA 証明書 (es-cert.p12) を含む Java™ トラストストア・ファイル。
オプション: Java トラストストア・ファイルのパスワード。

ブートストラップ・アドレスは、下記の画面上に外部リスナーとして表示されています。

したがって、kafkaproducer.propertiesは、以下のようになります。
kafkaproducer.propertiesとkafkaconsumer.propertiesは、/opt/ibm/InfoSphereDataReplication/ReplicationEngineforKafka/instance/ES/confのディレクトリに保管します。

bootstrap.servers=es-development-kafka-bootstrap-openshift-operators.apps.itz-g5ry1p.infra01-lb.dal14.techzone.ibm.com:443
security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="es-user" password="SCRAM_password";
ssl.truststore.location=/opt/ibm/InfoSphereDataReplication/ReplicationEngineforKafka/instance/ES/conf/es-cert.p12
ssl.truststore.password=truststore_password
ssl.truststore.type=PKCS12

以下のようになります。
JSONフォーマットでトピックにデータを連携するには、Kafkaカスタム操作プロセッサー(KCOP)の設定が必要です。
サブスクリプションを選択して、User Exitの設定を行います。
クラス名に下記を設定します。
com.datamirror.ts.target.publication.userexit.sample.kafka.KcopJsonFormatIntegrated
Start Refreshで全件送信してみます。
EMPLOYEEを送信します。
Monitoringタブに移り、右下のEventを開きます。
DB2TOESのサブスクリプションを選択して、Retrieve Eventを行うと、イベントの結果がでます。
ここでは、トピックが見つからないという事で終了しています。書き込みしようとしているトピックは、es.db2toes.sourcedb.db2admin.employee-jsonなので、これが存在しないのでエラーになっています。
トピックを作成して、再度試してみます。
トピックの作成ボタンでトピックを作成します。
トピック名を入力
パーティションの数を設定
メッセージは1日保存にします。
レプリカ設定はデフォルト
トピックが作成されました。
再度Refreshしてみます。
42レコードが送信されたログが出ています。
トピックの中を確認してみます。
確かに、Db2のデータが出ています。

念のため、DB2のデータ（EMPNO=00010)