Apache Flink CDC - Realtime Compute for Apache Flink - Alibaba Cloud ドキュメントセンター

このトピックでは、Realtime Compute for Apache Flink で Apache Flink Change Data Capture (CDC) コネクタを使用する方法と、その名前を変更する方法について説明します。

重要

Apache Flink CDC コネクタは、Apache 2.0 ライセンスに従うオープンソースコミュニティのプロジェクトです。そのサービスサポートとサービスレベルアグリーメント (SLA) は、Realtime Compute for Apache Flink に含まれる商用コネクタとは異なります。

構成の失敗、ジョブの失敗、データ損失などの問題が発生した場合、オープンソースコミュニティで解決策を見つける必要があります。Realtime Compute for Apache Flink チームは、Apache Flink CDC コネクタに関連する問題に対するテクニカルサポートを提供しません。
Apache Flink CDC コネクタを使用するサービスの SLA は、お客様の責任となります。

利用可能な CDC コネクタ

CDC コネクタ	説明
MySQL CDC PostgreSQL CDC (パブリックプレビュー) MongoDB CDC (パブリックプレビュー)	Realtime Compute for Apache Flink には、これらのコネクタの商用版が含まれています。そのため、オープンソースの Apache Flink CDC パッケージを手動でインストールする必要はありません。
OceanBase CDC Oracle CDC SQL Server CDC TiDB CDC Db2 CDC Vitess CDC	これらのコミュニティでサポートされている Apache Flink CDC コネクタの使用方法については、「コミュニティ CDC コネクタの使用」をご参照ください。説明 Apache Flink CDC コネクタまたはカスタムコネクタを使用する際、同じ名前の Realtime Compute for Apache Flink 組み込みコネクタまたはカスタムコネクタが既に存在する場合、組み込みコネクタとの競合を避けるために、デフォルトのコネクタ名を変更する必要があります。SQL Server CDC および Db2 CDC の場合、デフォルトのコミュニティコネクタ名を変更してから、コネクタを再パッケージ化する必要があります。たとえば、`sqlserver-cdc` を `sqlserver-cdc-test` に名前変更します。詳細については、「コネクタの名前変更」をご参照ください。

CDC と VVR のバージョンマッピング

VVR バージョン	CDC バージョン
vvr-6.0.7-flink-1.15	release-2.4
vvr-8.0.11-flink-1.17 ~ vvr-11.3-jdk11-flink-1.20	release-3.4
vvr-11.4-jdk11-flink-1.20+	release-3.5

コミュニティ CDC コネクタの使用

SQL ジョブ

Apache Flink CDC ページで、対象のコミュニティリリースバージョンをクリックします。最新の安定バージョンを使用することを推奨します。

説明
互換性の問題を避けるため、ご利用の VVR バージョンに対応するリリースバージョンを選択してください。バージョンマッピングについては、「CDC と VVR のバージョンマッピング」をご参照ください。
対象の CDC コネクタページで、JAR パッケージをダウンロードします。

SQL Client JAR セクションで、対応する JAR ファイルのダウンロードリンク (例: flink-sql-connector-mongodb-cdc-3.0.0.jar) をクリックします。

説明
対応する JAR ファイルは Maven Central Repository からもダウンロードできます。
Realtime Compute for Apache Flink コンソールにログインします。
対象のワークスペースの操作列で、コンソール をクリックします。
左側のナビゲーションウィンドウで、コネクタ をクリックします。
コネクタ ページで、カスタムコネクタの作成 をクリックし、ダウンロードした JAR ファイルをアップロードします。

詳細については、「カスタムコネクタの管理」をご参照ください。
SQL ジョブを作成する際、connector パラメーターの値として Apache Flink CDC コネクタの名前を使用します。

各コネクタがサポートするパラメーターについては、Apache Flink CDC ドキュメントをご参照ください。

JAR ジョブ

JAR ジョブで対応するコネクタを使用するには、pom.xml ファイルで次の依存関係を宣言する必要があります。
```
<dependency>
  <groupId>com.ververica</groupId>
  <artifactId>flink-connector-${the-connector-name}-cdc</artifactId>
  <version>${the-community-connector-version}</version>
</dependency>
```
Maven Central Repository にはリリースバージョンのみが含まれ、スナップショットバージョンは含まれません。スナップショットバージョンを使用するには、GitHub リポジトリをクローンし、JAR パッケージをコンパイルします。
コード内で、対応するコネクタの実装クラスを import し、ドキュメントの説明に従って使用します。
重要
アーティファクト ID が異なるアーティファクトの違いに注意してください： flink-connector-xxx と flink-sql-connector-xxx。
- flink-connector-xxx：コネクタのコードのみを含みます。他の依存関係はご自身で宣言する必要があります。
- flink-sql-connector-xxx：すべての依存関係を単一の JAR ファイルにバンドルしており、直接使用できます。
要件に基づいてアーティファクトを選択してください。たとえば、Flink 開発コンソールでカスタムコネクタを作成するには、flink-sql-connector-xxx を使用します。

コネクタの名前変更

このセクションでは、コミュニティの SQL Server CDC コネクタを例に、コネクタの名前を変更する方法を説明します。

GitHub リポジトリをクローンし、指定されたバージョンのブランチに切り替えます。

SQL Server CDC コネクタのファクトリークラスの識別子を変更します。

//com.ververica.cdc.connectors.sqlserver.table.SqlServerTableFactory
@Override
public String factoryIdentifier() {
    return "sqlserver-cdc-test";
}

flink-sql-connector-sqlserver-cdc サブモジュールをコンパイルしてパッケージ化します。
コネクタ ページで、カスタムコネクタの作成 をクリックし、パッケージ化された JAR ファイルをアップロードします。

詳細については、「カスタムコネクタの管理」をご参照ください。
SQL ジョブのステップに従って SQL ジョブを記述する際は、sqlserver-cdc-test パラメーターの値として、コネクタ名 connector を渡します。