ApsaraMQ for Kafka:SQL Server から Kafka Connect を使用した ApsaraMQ for Kafka へのデータ同期

データベースレベルでの CDC の有効化

USE testDB
GO
EXEC sys.sp_cdc_enable_db
GO

特定のテーブルの CDC の有効化

USE testDB
GO

EXEC sys.sp_cdc_enable_table
@source_schema = N'dbo',
@source_name   = N'MyTable',
@role_name     = N'MyRole',
@filegroup_name = N'MyDB_CT',
@supports_net_changes = 1
GO

パラメーターは次のとおりです。

CDC ステータスの確認

次のコマンドを実行して、CDC がアクティブであり、ご利用のアカウントが必要な権限を持っていることを確認してください。

EXEC sys.sp_cdc_help_change_data_capture
GO

結果が空の場合、ご利用のアカウントは CDC が有効なテーブルにアクセスできません。ロール割り当てを確認してください。

SQL Server Agent が実行中であることの確認

CDC は SQL Server Agent に依存します。次のコマンドを実行して、そのステータスを確認してください。

EXEC master.dbo.xp_servicecontrol N'QUERYSTATE',N'SQLSERVERAGENT'

出力が Running を示している場合、SQL Server Agent はアクティブです。

ApsaraMQ for Kafka での Topic の作成

コネクタを開始する前に、ApsaraMQ for Kafka コンソールで必要な Topic を作成してください。コネクタは、<server-name>.<database>.<table> という名前の Topic に変更イベントを書き込みます。

このチュートリアルのテストデータベース (testDB に 4 つのテーブル) の場合、次の Topic を作成してください。

Topic の作成方法については、「Topic の作成」をご参照ください。または、CreateTopic API を呼び出します。

コネクタの構成

1. register-sqlserver.json をダウンロードします。

2. register-sqlserver.json を開き、アクセス方法に基づいて次のプロパティを更新します。VPC アクセス インターネットアクセス 主要なプロパティは次のとおりです。インターネットアクセスの場合、追加の SSL および SASL プロパティにより、ApsaraMQ for Kafka への暗号化された認証済み接続が可能になります。

   プロパティ	説明
   database.history.kafka.bootstrap.servers	ご利用の ApsaraMQ for Kafka インスタンスのエンドポイント。VPC アクセスにはデフォルトエンドポイントを、インターネットアクセスには SSL エンドポイントを使用します。エンドポイントは ApsaraMQ for Kafka コンソールで確認できます。
   database.server.name	すべての変更イベント Topic のプレフィックスとして使用される論理サーバー名。server1 に設定すると、server1.testDB.products のような Topic が生成されます。
   database.history.kafka.topic	コネクタがスキーマ変更履歴を保存する Topic。コネクタを開始する前に、この Topic をコンソールで作成してください。

   "database.history.kafka.bootstrap.servers" : "<your-default-endpoint>",
   "database.server.name": "server1",
   "database.history.kafka.topic": "schema-changes-inventory"

   "database.history.kafka.bootstrap.servers" : "<your-ssl-endpoint>",
   "database.server.name": "server1",
   "database.history.kafka.topic": "schema-changes-inventory",
   "database.history.producer.ssl.truststore.location": "kafka.client.truststore.jks",
   "database.history.producer.ssl.truststore.password": "KafkaOnsClient",
   "database.history.producer.security.protocol": "SASL_SSL",
   "database.history.producer.sasl.mechanism": "PLAIN",
   "database.history.consumer.ssl.truststore.location": "kafka.client.truststore.jks",
   "database.history.consumer.ssl.truststore.password": "KafkaOnsClient",
   "database.history.consumer.security.protocol": "SASL_SSL",
   "database.history.consumer.sasl.mechanism": "PLAIN"

コネクタの開始

POST リクエストを送信して、Kafka Connect にコネクタを登録します。

curl -i -X POST \
  -H "Accept:application/json" \
  -H "Content-Type:application/json" \
  http://localhost:8083/connectors/ \
  -d @register-sqlserver.json