Data Transmission Service:Kafkaクライアントを使用して追跡データを消費する

使用上の注意

• 変更追跡機能を使用するときに自動コミットを有効にすると、一部のデータが消費される前にコミットされることがあります。 これにより、データが失われます。 手動でデータをコミットすることを推奨します。

  説明

  データのコミットに失敗した場合は、クライアントを再起動して、最後に記録された消費チェックポイントからデータを消費し続けることができます。 しかしながら、この期間中に重複データが生成されることがある。 重複データを手動で除外する必要があります。

• データはシリアル化され、Avro形式で保存されます。 詳細については、「Record.avsc」をご参照ください。

  警告

  このトピックで説明しているKafkaクライアントを使用していない場合は、Avroスキーマに基づいて追跡データを解析する必要があります。

• 検索単位は、データ送信サービス (DTS) がoffsetForTimes操作を呼び出すときに2番目になります。 ネイティブKafkaクライアントがこの操作を呼び出す場合、検索単位はミリ秒です。

• 障害復旧などのいくつかの理由により、Kafkaクライアントと変更追跡サーバーの間で一時的な接続が発生する場合があります。 このトピックで説明しているKafkaクライアントを使用していない場合は、Kafkaクライアントにネットワーク再接続機能が必要です。

Kafkaクライアントを実行する

Kafka clientデモをダウンロードします。 デモの使用方法の詳細については、「Readme」をご参照ください。

手順

次の手順は、Kafkaクライアントを実行して追跡データを消費する方法を示しています。 この例では、IntelliJ IDEA Community Edition 2018.1.4 for Windowsが使用されています。

1. 変更追跡タスクを作成します。 詳細については、「ApsaraDB RDS For MySQLインスタンスからのデータ変更の追跡」、「PolarDB for MySQLクラスターからのデータ変更の追跡」、または「自己管理型Oracleデータベースからのデータ変更の追跡」をご参照ください。

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

3. Kafkaクライアントデモをダウンロードし、パッケージを解凍します。

   説明

   アイコンをクリックし、[ZIPのダウンロード] を選択してパッケージをダウンロードします。

4. IntelliJ IDEAを開きます。 表示されるウィンドウで、[開く] をクリックします。

   

5. 表示されるダイアログボックスで、ダウンロードしたデモが存在するディレクトリに移動します。 pom.xmlファイルを見つけます。

   

6. 表示されるダイアログボックスで、[プロジェクトとして開く] を選択します。

7. IntelliJ IDEAのプロジェクトツールウィンドウで、フォルダーをクリックしてKafkaクライアントのデモファイルを検索し、ファイルをダブルクリックします。 ファイル名はNotifyDemoDB.javaです。

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

   

   パラメーター	説明	パラメータ値を取得するメソッド
   USER_NAME	消費者グループのアカウント。 警告 このトピックで説明するKafkaクライアントを使用していない場合は、<Username>-<Consumer group ID> の形式でアカウントを指定する必要があります。 例: dtstest-dtsae ****** bpv そうしないと、接続は失敗します。	DTSコンソールで、管理する変更追跡インスタンスを見つけ、インスタンスIDをクリックします。 左側のナビゲーションウィンドウで、[データの使用] をクリックします。 表示されるページで、コンシューマーグループIDと対応するユーザー名を取得できます。 説明 コンシューマーグループアカウントのパスワードは、コンシューマーグループの作成時に自動的に指定されます。
   パスワード名	アカウントのパスワードを入力します。
   SID_NAME	コンシューマーグループの ID です。
   グループ名	消費者グループの名前です。 このパラメーターをコンシューマーグループIDに設定します。
   KAFKA_TOPIC	変更追跡インスタンスのトピック。	DTSコンソールで、管理する変更追跡インスタンスを見つけ、インスタンスIDをクリックします。 [タスク管理] ページで、トピックとネットワーク情報を取得できます。
   KAFKA_BROKER_URL_NAME	変更追跡インスタンスのエンドポイント。 説明 内部ネットワーク上のデータ変更を追跡する場合、ネットワーク遅延は最小限に抑えられます。 これは、KafkaクライアントをデプロイするElastic Compute Service (ECS) インスタンスがクラシックネットワーク上にある場合、または変更追跡インスタンスと同じ仮想プライベートクラウド (VPC) にある場合に適用されます。
   INITIAL_CHECKPOINT_NAME	消費されたデータの消費チェックポイント。 この値は UNIX タイムスタンプです。 例: 1592269238。 説明 次の理由で、消費チェックポイントを保存する必要があります。 消費プロセスが中断された場合、Kafkaクライアントで消費チェックポイントを指定してデータ消費を再開できます。 これにより、データの損失を防ぎます。 Kafkaクライアントを起動すると、オンデマンドでデータを消費するための消費チェックポイントを指定できます。 SUBSCRIBE_MODE_NAMEパラメーターがsubscribeに設定されている場合、指定したINITIAL_CHECKPOINT_NAMEパラメーターは、Kafkaクライアントを初めて起動したときにのみ有効になります。	次の図に示すように、消費されるデータの消費チェックポイントは、変更追跡インスタンスのデータ範囲内にある必要があります。 消費チェックポイントは、UNIXタイムスタンプに変換する必要があります。 説明 検索エンジンを使用して、UNIXタイムスタンプコンバーターを取得できます。
   USE_CONFIG_CHECKPOINT_NAME	指定された消費チェックポイントからのデータをクライアントに消費させるかどうかを指定します。 デフォルト値: true。 このパラメーターをtrueに設定すると、受信したが処理されていないデータが失われないようにすることができます。	非該当
   SUBSCRIBE_MODE_NAME	コンシューマグループに対して2つ以上のKafkaクライアントを実行するかどうかを指定します。 この機能を使用する場合は、このパラメーターをこれらのKafkaクライアントのサブスクライブに設定します。 デフォルト値はassignで、この機能が使用されていないことを示します。 コンシューマグループに1つのKafkaクライアントのみをデプロイすることを推奨します。	非該当

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

   説明

   IntelliJ IDEAを初めて実行する場合、関連する依存関係をロードしてインストールするには、特定の期間が必要です。

Kafkaクライアントの結果

次の図は、Kafkaクライアントがソースデータベースからのデータ変更を追跡できることを示しています。

NotifyDemoDB.javaファイルの25行目の // log.info(ret); 文字列から2重スラッシュ (//) を削除できます。 次に、クライアントを再度実行して、データ変更情報を表示します。

よくある質問

• Q: Kafkaクライアントの消費チェックポイントを記録する必要があるのはなぜですか?

  A: DTSによって記録された消費チェックポイントは、DTSがKafkaクライアントからコミット操作を受信した時点です。 記録された消費チェックポイントは、実際の消費時間とは異なり得る。 ビジネスアプリケーションまたはKafkaクライアントが予期せず中断された場合は、正確な消費チェックポイントを指定してデータの消費を継続できます。 これにより、データの損失や重複データの消費を防ぎます。

MySQLデータ型とdataTypeNumber値のマッピング

Oracleデータ型とdataTypeNumber値のマッピング

PostgreSQLデータ型とdataTypeNumber値のマッピング