SeaTunnelは、分散型で高い拡張性を備えたデータ統合プラットフォームであり、さまざまなデータストレージシステム間でデータを転送するために使用されます。SeaTunnelは設定が容易なデータストリームタスクプラグインを提供し、ユーザーがデータ処理と変換のプロセスを定義できるようにします。SeaTunnelの詳細については、SeaTunnelのGitHubアドレスをご参照ください。
前提条件
データを同期する前に、以下の事項を確認してください：
- SeaTunnelをダウンロード済みであること。
- Java 1.8以降のバージョンがインストールされていること。
- OB Cloud MySQLのBinlogが有効になっていること。
- MySQLドライバーを
libディレクトリにダウンロード済みであること。
SeaTunnelのMySQL CDC Source Connectorを使用して、OB Cloud MySQL互換モードのテナントでSQLクエリを実行する方法について説明します。connector-cdc-mysqlはSeaTunnelのプラグインであり、MySQLデータベース内のリアルタイム変更データ（Change Data Capture、CDC）をキャプチャし、ストリーミングで転送するために使用されます。
操作手順
以下の手順に従って、OB Cloud MySQL互換モードのテナント間でデータを同期します：
OB CloudのBinlogサービスを有効にします。
Binlogログサービスを有効にするパス：インスタンスリスト -> テナント管理 -> Binlogサービスで、「開始」をクリックします。詳細については、Binlogログサービスの開始をご参照ください。
SeaTunnelユーザーを作成します。
CREATE USER 'ユーザー名'@'%' IDENTIFIED BY 'パスワード'; GRANT SELECT ON *.* TO 'ユーザー名';config/plugin_configを設定します。connector-cdc-mysqlを指定し、sh bin/install-plugin.sh 2.3.3を実行します。SeaTunnelを設定します。
設定ファイル
config/v2.streaming.conf.templateを修正します：source { # これは、**機能のテストおよびデモンストレーションのみを目的とした** ソースプラグインの例です。 MySQL-CDC { result_table_name = "fake" parallelism = 1 server-id = 5656 username = "ユーザー名" password = "パスワード" table-names = ["seatunnel.full_types"] base-url = "jdbc:mysql://xx.xxx.x.xx:3306/seatunnel" } # SeaTunnelの設定方法に関する詳細情報やソースプラグインの全リストについては、 # https://seatunnel.apache.org/docs/category/source-v2 をご参照ください。 }SeaTunnelを起動します。
以下のコマンドを実行して、SeaTunnelを起動します：
./bin/seatunnel.sh --config ./config/v2.streaming.conf.template -e local
注意
OB CloudはSET型をサポートしていません。
Caused by: org.apache.seatunnel.api.table.catalog.DataTypeConvertException: ErrorCode:[COMMON-07], ErrorDescription:[Unsupported data type] - Convert type: SET to SeaTunnel data type error.
詳細については、SeaTunnel MySQL CDCドキュメントをご参照ください。
SeaTunnelのJDBC OceanBase Source ConnectorとJDBC OceanBase Sink Connectorを使用したデータ同期について説明します。JDBC OceanBase Source Connectorは、OceanBaseの単一テーブルデータのフル読み取りと書き込みをサポートします。詳細については、JDBC OceanBase Source Connectorをご参照ください。JDBC OceanBase Sink Connectorは、OceanBaseの複数テーブルデータのフル読み取りと書き込みをサポートします。詳細については、JDBC OceanBase Sink Connectorをご参照ください。
前提条件
データを同期する前に、以下の事項を確認してください：
- SeaTunnelをダウンロード済みであること。
- Java 1.8以降をインストール済みであること。
config/plugin_configが設定されていること。詳細については、操作手順をご参照ください。- OB Cloud MySQLのBinlogが有効になっていること。詳細については、操作手順をご参照ください。
- ターゲットのOB Cloudデータベースに、ソースデータベースと同じテーブル構造を作成済みであること。
操作手順
以下の手順に従って、OB CloudのMySQL互換モードのテナント間でデータを同期します：
OB CloudのBinlogサービスを有効にします。
Binlogログサービスを有効にするパス：インスタンスリスト -> テナント管理 -> Binlogサービスで、「開始」をクリックします。詳細については、Binlogログサービスの開始をご参照ください。
SeaTunnelユーザーを作成します。
CREATE USER 'ユーザー名'@'%' IDENTIFIED BY 'パスワード'; GRANT SELECT, INSERT, UPDATE, DELETE ON *.* TO 'ユーザー名';config/plugin_configを設定します。OceanBase JDBCドライバーを
libディレクトリにダウンロードします。詳細については、OceanBase Clientをご参照ください。SeaTunnelを設定します。
設定ファイル
config/ob_to_obを修正します：env { parallelism = 1 job.mode = "BATCH" } source { # This is a example source plugin **only for test and demonstrate the feature source plugin** Jdbc { driver = com.oceanbase.jdbc.Driver url = "jdbc:oceanbase://e2e_oceanbase_mysql:3306/seatunnel?useUnicode=true&characterEncoding=UTF-8&rewriteBatchedStatements=true&serverTimezone=UTC" user = "ユーザー名" password = "パスワード" query = "SELECT c_bit_1, c_bit_8 FROM source" compatible_mode = "mysql" } # If you would like to get more information about how to configure seatunnel and see full list of source plugins, # please go to https://seatunnel.apache.org/docs/connector-v2/source/FakeSource } sink { Jdbc { driver = com.oceanbase.jdbc.Driver url = "jdbc:oceanbase://e2e_oceanbase_mysql:3306/seatunnel?useUnicode=true&characterEncoding=UTF-8&rewriteBatchedStatements=true&serverTimezone=UTC" user = "ユーザー名" password = "パスワード" query = "insert into sink(c_bit_1, c_bit_8) values (?, ?);" compatible_mode = "mysql" } # If you would like to get more information about how to configure seatunnel and see full list of sink plugins, # please go to https://seatunnel.apache.org/docs/connector-v2/sink }SeaTunnelを起動します。
以下のコマンドを実行して、SeaTunnelを起動します：
./bin/seatunnel.sh --config ./config/ob_to_ob -e local
注意
OB CloudはSET型をサポートしていません。
Caused by: org.apache.seatunnel.api.table.catalog.DataTypeConvertException: ErrorCode:[COMMON-07], ErrorDescription:[ Unsupported data type] - Convert type: SET to SeaTunnel data type error.