Kafkaは、高スループットと高スケーラビリティを備えた分散メッセージキューサービスです。 Kafkaは、ログ収集、モニタリングデータ集約、ストリーミング処理、オンラインおよびオフライン分析などのビッグデータ分析に広く使用されています。 ビッグデータエコシステムにとって重要です。 このトピックでは、data Transmission Service (DTS) を使用して、ApsaraDB RDS for MySQLインスタンスから自己管理型Kafkaクラスターにデータを移行する方法について説明します。 自己管理KafkaクラスターにはパブリックIPアドレスがあります。 データ移行機能を使用すると、メッセージ処理機能を拡張できます。
前提条件
Kafkaクラスターが作成され、Kafkaのバージョンは0.10.1.0〜2.7.0です。
Kafkaクラスターのサービスポートは、インターネット経由でアクセスできる必要があります。
背景情報
データ同期機能を使用する場合、自己管理Kafkaクラスターのデプロイは、次のいずれかのタイプに属している必要があります。
Elastic Compute Service (ECS) でホストされる自己管理型データベース
Express Connect、VPN Gateway、またはSmart Access Gatewayで接続された自己管理型データベース
パブリックIPアドレスまたはポート番号のない自己管理データベース (database Gateway経由で接続)
Cloud Enterprise Network (CEN) 経由で接続された自己管理型データベース
自己管理Kafkaクラスターのデプロイが上記のタイプのいずれにも属していない場合は、Kafkaクラスターのサービスポートをインターネットに開くことができます。 次に、データ移行機能を使用して、データをKafkaクラスターに同期できます。
注意事項
DTSは、完全データ移行中にソースデータベースとターゲットデータベースの読み取りおよび書き込みリソースを使用します。 これは、データベースサーバの負荷を増加させる可能性がある。 データベースのパフォーマンスが悪い場合、仕様が低い場合、またはデータ量が多い場合、データベースサービスが利用できなくなる可能性があります。 たとえば、ソースデータベースで多数の低速SQLクエリが実行されている場合、テーブルにプライマリキーがない場合、またはターゲットデータベースでデッドロックが発生する場合、DTSは大量の読み取りおよび書き込みリソースを占有します。 データを移行する前に、移行元データベースと移行先データベースのパフォーマンスに対するデータ移行の影響を評価します。 オフピーク時にデータを移行することを推奨します。 たとえば、ソースデータベースとターゲットデータベースのCPU使用率が30% 未満の場合にデータを移行できます。
ソースデータベースにはPRIMARY KEYまたはUNIQUE制約が必要で、すべてのフィールドが一意である必要があります。 そうでない場合、宛先データベースは重複するデータレコードを含み得る。
移行するオブジェクトとしてテーブルのみを選択できます。
課金ルール
移行タイプ | タスク設定料金 | インターネットトラフィック料金 |
スキーマ移行とフルデータ移行 | 無料です。 | インターネット経由でAlibaba Cloudからデータが移行された場合にのみ課金されます。 詳細については、「課金の概要」をご参照ください。 |
増分データ移行 | 有料。 詳細については、「課金の概要」をご参照ください。 |
手順
にログインします。 DTSコンソール。
説明データ管理 (DMS) コンソールにリダイレクトされている場合は、にある
アイコンをクリックして、以前のバージョンのDTSコンソールに移動し
。左側のナビゲーションウィンドウで、[データ移行] をクリックします。
[移行タスク] ページの上部で、移行先クラスターが存在するリージョンを選択します。
ページの右上隅にある [移行タスクの作成] をクリックします。
ソースデータベースとターゲットデータベースを設定します。
セクション
パラメーター
説明
非該当
タスク名
DTSが自動的に生成するタスク名。 タスクを簡単に識別できるように、わかりやすい名前を指定することをお勧めします。 一意のタスク名を指定する必要はありません。
移行元データベース
インスタンスタイプ
RDS インスタンスを選択します。
インスタンスリージョン
ソースApsaraDB RDSインスタンスが存在するリージョン。
インスタンス ID
ソースApsaraDB RDSインスタンスのID。
データベースアカウント
ソースインスタンスのデータベースアカウント。 アカウントには、移行するオブジェクトに対するSELECT権限と、REPLICATION CLIENT、REPLICATION SLAVE、およびSHOW VIEW権限が必要です。
データベースパスワード
データベースアカウントのパスワードを設定します。
暗号化
ソースインスタンスへの接続を暗号化するかどうかを指定します。 [暗号化なし] または[SSL 暗号化] を選択します。 SSL暗号化を選択した場合、データ移行タスクを設定する前に、ApsaraDB RDS for MySQLインスタンスのSSL暗号化を有効にする必要があります。 詳細については、「ApsaraDB RDS For MySQLインスタンスのSSL暗号化の設定」をご参照ください。
重要Encryptionパラメーターは、中国本土および中国 (香港) リージョン内でのみ使用できます。
ターゲットデータベース
インスタンスタイプ
[オンプレミスデータベース] を選択します。
Instance Region
指定する必要はありません。
データベースエンジン
Kafkaを選択します。
Hostname or IP Address
自己管理Kafkaクラスターへのアクセスに使用されるIPアドレス。 この例では、パブリックIPアドレスが使用されます。
ポート番号
Kafkaクラスターのサービスポート番号。 デフォルト値: 9092
データベースアカウント
Kafkaクラスターへのログインに使用されるユーザー名。 Kafkaクラスターで認証が有効になっていない場合は、ユーザー名を入力する必要はありません。
データベースパスワード
ユーザー名のパスワード。 Kafkaクラスターで認証が有効になっていない場合は、パスワードを入力する必要はありません。
トピック
[トピックリストの取得] をクリックし、ドロップダウンリストからトピック名を選択します。
Kafkaバージョン
ターゲットKafkaクラスターのバージョン。
暗号化
接続先クラスターへの接続を暗号化するかどうかを指定します。 ビジネスとセキュリティの要件に基づいて、[非暗号化] または [SCRAM-SHA 256] を選択します。
ページの右下隅にある [ホワイトリストと次への設定] をクリックします。
警告DTSサーバーのCIDRブロックがデータベースまたはインスタンスのホワイトリスト、またはECSセキュリティグループルールに自動的または手動で追加されると、セキュリティリスクが発生する可能性があります。 したがって、DTSを使用してデータを移行する前に、潜在的なリスクを理解して認識し、次の対策を含む予防策を講じる必要があります。VPNゲートウェイ、またはSmart Access Gateway。
移行タイプ、移行ポリシー、および移行するオブジェクトを選択します。
設定
説明
移行タイプの選択
[スキーマ移行] 、[フルデータ移行] 、および [増分データ移行] を選択します。
重要増分データ移行が選択されていない場合、完全データ移行中にソースデータベースにデータを書き込まないことを推奨します。 これにより、ソースデータベースとターゲットデータベース間のデータの整合性が確保されます。
Kafkaで使用するデータ形式を選択する
Kafkaクラスターに移行されたデータは、AvroまたはCanal JSON形式で保存されます。 詳細については、「Kafkaクラスターのデータ形式」をご参照ください。
データをKafkaパーティションに移行するためのポリシーを選択する
ビジネス要件に基づいて移行ポリシーを選択します。 詳細については、「Kafkaパーティションにデータを同期するためのポリシーの指定」をご参照ください。
移行するオブジェクトを選択します。
[使用可能] セクションから1つ以上のテーブルを選択し、
アイコンをクリックして、[選択済み] セクションにテーブルを追加します。 説明DTSは、手順5で選択したトピック名にテーブル名をマップします。 トピックの名前を変更する方法については、「オブジェクト名マッピング」をご参照ください。
オブジェクトの名前を変更するかどうかを指定する
オブジェクト名マッピング機能を使用して、移行先インスタンスに移行されるオブジェクトの名前を変更できます。 詳細は、オブジェクト名のマッピングをご参照ください。
ソースデータベースまたはターゲットデータベースへの接続が失敗した場合のリトライ時間範囲の指定
デフォルトでは、DTSがソースデータベースまたはターゲットデータベースへの接続に失敗した場合、DTSは次の12時間以内に再試行します。 業務要件に基づいて再試行時間範囲を指定できます。 指定された時間範囲内にDTSがソースデータベースとターゲットデータベースに再接続された場合、DTSはデータ移行タスクを再開します。 それ以外の場合、データ移行タスクは失敗します。
説明DTSがソースデータベースとターゲットデータベースへの再接続を試みる時間範囲内で、DTSインスタンスに対して課金されます。 業務要件に基づいて再試行時間範囲を指定することを推奨します。 ソースデータベースとターゲットデータベースがリリースされた後、できるだけ早くDTSインスタンスをリリースすることもできます。
DMSがソーステーブルに対してオンラインDDL操作を実行するときに、一時テーブルをターゲットデータベースにコピーするかどうかを指定します。
Data Management (DMS) を使用してソースデータベースでオンラインDDL操作を実行する場合、オンラインDDL操作によって生成された一時テーブルを移行するかどうかを指定できます。
はい: DTSは、オンラインDDL操作によって生成された一時テーブルのデータを移行します。
説明オンラインDDL操作で大量のデータが生成される場合、移行タスクの待ち時間が発生する可能性があります。
No: DTSは、オンラインDDL操作によって生成された一時テーブルのデータを移行しません。 移行元データベースの元のDDLデータのみが移行されます。
説明[いいえ] を選択すると、ターゲットデータベースのテーブルがロックされる可能性があります。
ページの右下隅にある [事前チェック] をクリックします。
説明データ移行タスクを開始する前に、DTSは事前チェックを実行します。 データ移行タスクは、タスクが事前チェックに合格した後にのみ開始できます。
タスクが事前チェックに合格しなかった場合は、失敗した各項目の横にある
アイコンをクリックして詳細を表示できます。 原因に基づいて問題をトラブルシューティングし、事前チェックを再度実行できます。
問題をトラブルシューティングする必要がない場合は、失敗した項目を無視して、再度事前チェックを実行できます。
タスクが事前チェックに合格したら、[次へ] をクリックします。
[設定の確認] ダイアログボックスで、[チャネル仕様] パラメーターを指定し、[データ送信サービス (従量課金) サービス規約] を選択します。
[購入と開始] をクリックして、データ移行タスクを開始します。
スキーマ移行とフルデータ移行
フルデータ移行中は、手動でタスクを停止しないことをお勧めします。 そうしないと、ターゲットデータベースに移行されたデータが不完全になる可能性があります。 データ移行タスクが自動的に停止するまで待つことができます。
スキーマ移行、完全データ移行、および増分データ移行
増分データ移行タスクは自動的に停止しません。 タスクを手動で停止する必要があります。
重要データ移行タスクを手動で停止する適切な時期を選択することを推奨します。 たとえば、オフピーク時やワークロードを移行先クラスターに切り替える前にタスクを停止できます。
増分データ移行と移行タスクが遅延しませんが移行タスクのプログレスバーに表示されるまで待ちます。 その後、ソースデータベースへのデータの書き込みを数分間停止します。 増分データ移行のレイテンシは、プログレスバーに表示され得る。
増分データ移行のステータスが [移行タスクは遅延なし] に変わるまで待ちます。 次に、移行タスクを手動で停止します。
