Data Transmission Service:Kafka クライアントによるサブスクライブデータの消費

注意事項

• このトピックで提供されるデモを使用して自動コミットを有効にすると、すべてのデータが消費される前にコミット操作が実行される可能性があるため、一部のデータが失われることがあります。この問題を回避するには、手動コミットを使用してください。

  説明

  障害によりコミットが失敗した場合、再起動後、クライアントは最後に記録されたオフセットからデータの消費を再開します。これにより、データが重複する可能性があります。重複したデータは手動でフィルターしてください。

• データは Avro シリアル化フォーマットで保存されます。フォーマットの詳細については、Record.avsc ドキュメントをご参照ください。

  警告

  このトピックで提供されている Kafka クライアントを使用しない場合、デシリアライゼーション (DTS Avro デシリアライゼーションの例をご参照ください) で不正な結果になる可能性があります。ご自身でデータの正確性を検証してください。

• Data Transmission Service (DTS) は、offsetForTimes API の検索単位として秒を使用しますが、ネイティブ Kafka はミリ秒を使用します。

• ディザスタリカバリなどの理由により、変更追跡サーバーでネットワークの一時的な切断が発生する場合があります。このトピックで提供されている Kafka クライアントを使用しない場合は、お使いの Kafka クライアントがネットワークリトライをサポートしていることを確認してください。

• ネイティブ Kafka クライアントを使用してサブスクライブしたデータを消費する場合、DTS が増分データインジェストモジュールを切り替えることがあります。これにより、サブスクライブモードでサーバーに保存されているコンシューマーオフセットがクリアされます。必要に応じてコンシューマーオフセットを手動で調整し、データを消費してください。

Kafka クライアントのワークフロー

Kafka クライアントのデモコードをダウンロードします。コードの使用方法の詳細については、デモ内の Readme ファイルをご参照ください。

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.0.0</version>
</dependency>

操作手順

このトピックでは、IntelliJ IDEA (Community Edition 2018.1.4 for Windows) を例に、クライアントを実行して変更追跡チャネルからデータを消費する方法について説明します。

1. 新しい変更追跡チャネルを作成します。詳細については、「RDS MySQL 変更追跡チャネルの作成」、「PolarDB MySQL 変更追跡チャネルの作成」、または「Oracle 変更追跡チャネルの作成」をご参照ください。

2. 1 つ以上のコンシューマーグループを作成します。詳細については、「コンシューマーグループの追加」をご参照ください。

3. Kafka クライアントのデモコード をダウンロードし、ファイルを展開します。

   説明

   をクリックし、[Download ZIP] を選択してファイルをダウンロードします。

4. IntelliJ IDEA を開き、[Open] をクリックします。

5. 表示されたダイアログボックスで、Kafka クライアントのデモコードをダウンロードしたディレクトリに移動します。フォルダーをたどって、プロジェクトオブジェクトモデルファイル pom.xml を見つけます。

   kafkademo > subscribe_example-master > javaimpl に移動し、pom.xml を選択して、[OK] をクリックします。

6. 表示されたダイアログボックスで、[Open as Project] を選択します。

7. IntelliJ IDEA のインターフェイスで、フォルダーをたどって Kafka クライアントのデモファイル NotifyDemoDB.java を見つけ、ダブルクリックします。

8. NotifyDemoDB.java ファイルのパラメーター値を設定します。

   public static Properties getConfigs() {
       Properties properties = new Properties();
       // 認証用のユーザー名、パスワード、SID
       properties.setProperty(USER_NAME, "dtstest");
       properties.setProperty(PASSWORD_NAME, "xxx");
       properties.setProperty(SID_NAME, "dtsxxx");
       // Kafka コンシューマーグループ。通常は SID と同じです。
       properties.setProperty(GROUP_NAME, "dtsxxx");
       // 消費するトピック。パーティションは 0 です。
       properties.setProperty(KAFKA_TOPIC, "cn_hangzhou_xxx");
       // Kafka ブローカーの URL
       properties.setProperty(KAFKA_BROKER_URL_NAME, "dts-cn-xxx.com:18001");
       // 最初のシーク用の初期チェックポイント (設定するタイムスタンプ。例: 1566180200 (2019年8月19日(月) 10:03:21 CST の場合))
       properties.setProperty(INITIAL_CHECKPOINT_NAME, "1583307907");
       // 開始時に設定チェックポイントを強制的に使用するかどうか。チェックポイントのリセットに使用します。
       properties.setProperty(USE_CONFIG_CHECKPOINT_NAME, "true");
       // consumer の assign または subscribe インターフェイスを使用します。
       // subscribe モードを使用する場合、グループ設定が必要です。これにより Kafka コンシューマーグループが有効になります。
       properties.setProperty(SUBSCRIBE_MODE_NAME, "assign");
       return properties;
   }

   パラメーター	説明	取得方法
   USER_NAME	コンシューマーグループのアカウントです。 警告 このトピックで提供されているクライアントを使用しない場合は、ユーザー名を <コンシューマーグループアカウント>-<コンシューマーグループ ID> の形式 (例: dtstest-dtsae******bpv) で設定してください。そうしないと、接続は失敗します。	DTS コンソールで、対象のサブスクリプションインスタンス ID をクリックし、データ消費 をクリックします。[コンシューマーグループ ID] と アカウント の情報を取得できます。 説明 コンシューマーグループアカウントのパスワードは、コンシューマーグループの作成時に指定したものです。
   PASSWORD_NAME	アカウントのパスワードです。
   SID_NAME	コンシューマーグループ ID です。
   GROUP_NAME	コンシューマーグループ名です。このパラメーターをコンシューマーグループ ID と同じ値に設定します。
   KAFKA_TOPIC	変更追跡チャネルのサブスクリプショントピックです。	DTS コンソールで、対象のサブスクリプションインスタンス ID をクリックします。タスク管理 ページでは、トピック とネットワークアドレス情報を取得できます。[基本情報] セクションから [サブスクリプショントピック] の値、および [ネットワーク] セクションから [VPC エンドポイント] (形式例: xxx.aliyuncs.com:18003) を取得します。
   KAFKA_BROKER_URL_NAME	データサブスクリプションチャネルのブローカー URL です。 説明 Kafka クライアントをデプロイする ECS インスタンスと変更追跡チャネルがクラシックネットワークまたは同じ VPC 内にある場合は、内部エンドポイントを使用するとネットワークレイテンシを最小限に抑えられます。 ネットワークが不安定になる可能性があるため、パブリックエンドポイントの使用は推奨されません。
   INITIAL_CHECKPOINT_NAME	データ消費を開始するタイムスタンプです。UNIX タイムスタンプ形式 (例: 1592269238) で指定します。 説明 以下の操作を行うために、タイムスタンプをご自身で保存してください。 プログラムが中断された後、最後に消費されたタイムスタンプからデータ消費を再開することで、データ損失を防ぎます。 サブスクリプションクライアントの起動時に目的のコンシューマーオフセットを指定して、必要に応じてデータを消費します。 SUBSCRIBE_MODE_NAME が subscribe に設定されている場合、INITIAL_CHECKPOINT_NAME の値は、サブスクリプションクライアントの初回起動時にのみ有効になります。	データタイムスタンプは、サブスクリプションインスタンスのタイムスタンプ範囲内であり、かつ UNIX タイムスタンプに変換されている必要があります。DTS のサブスクリプションタスクリストで、サブスクリプションタスクの [タイムスタンプ範囲] フィールドを確認し、INITIAL_CHECKPOINT_NAME の有効な範囲を特定します。 説明 検索エンジンを使用して、UNIX タイムスタンプコンバーターを検索してください。
   USE_CONFIG_CHECKPOINT_NAME	デフォルト値は true です。この設定により、指定されたデータタイムスタンプからの消費が強制されるため、受信済みで未処理のデータの損失を防ぎます。	なし
   SUBSCRIBE_MODE_NAME	1 つのコンシューマーグループで 2 つ以上の Kafka クライアントを実行する場合は、このパラメーターをすべてのクライアントで subscribe に設定してください。 デフォルト値は assign です。この場合、この機能は無効になります。このモードでは、クライアントを 1 つだけデプロイしてください。	なし

9. IntelliJ IDEA のインターフェイス上部にあるメニューで、[Run] > [Run] を選択してクライアントを実行します。

   説明

   初回実行時、ソフトウェアが依存関係を自動的にロードおよびインストールするため、時間がかかることがあります。

実行結果

クライアントを実行すると、ソースデータベースからのデータ変更を正常にサブスクライブします。

[2020-03-09 10:41:52,408] INFO [Consumer clientId=consumer-1, groupId=dts_xxx] Discovered coordinator xxx (id: xxx rack: null) (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
[2020-03-09 10:41:57,203] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721711, offset: 1732521, info: 1583721711] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:41:57,571] INFO EtlRecordProcessor: haven't receive records from generator for  5s (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:02,203] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721721, offset: 1732539, info: 1583721721] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:07,204] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721726, offset: 1732544, info: 1583721726] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:12,205] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721731, offset: 1732548, info: 1583721731] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:17,205] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721736, offset: 1732554, info: 1583721736] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:22,205] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721741, offset: 1732559, info: 1583721741] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:27,206] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721746, offset: 1732569, info: 1583721746] (recordprocessor.EtlRecordProcessor)

また、NotifyDemoDB.java ファイル内のログ出力行のコメントを解除し (25 行目の //log.info(ret); から // を削除)、クライアントを再度実行することで、詳細なデータ変更情報を確認することもできます。

よくある質問

• Q： コンシューマーオフセットを自分で記録する必要があるのはなぜですか。

  A： DTS は、Kafka コンシューマークライアントからコミット操作を受信した時点に基づいてコンシューマーオフセットを記録します。この時刻は、実際の消費時刻と異なる場合があります。アプリケーションまたは Kafka コンシューマークライアントが異常中断した後は、お客様が記録したオフセットから消費を再開することで、データの重複や欠損を回避できます。

MySQL フィールドタイプと dataTypeNumber 値のマッピング

Oracle フィールドタイプと dataTypeNumber 値のマッピング

PostgreSQL のフィールドタイプと dataTypeNumber 値の対応