AnalyticDB for MySQL は、AnalyticDB パイプラインサービス (APS) 機能を提供します。この機能を使用すると、Kafka 同期タスクを作成して、指定されたオフセットから Kafka のデータをリアルタイムでデータインジェストできます。このトピックでは、Kafka データソースの追加、Kafka 同期タスクの作成、およびタスクの開始方法について説明します。
前提条件
AnalyticDB for MySQL の Enterprise Edition、Basic Edition、または Data Lakehouse Edition クラスターが作成されていること。
AnalyticDB for MySQL クラスター用のデータベースアカウントが作成されていること。
Alibaba Cloud アカウントを使用する場合は、特権アカウントを作成するだけで済みます。
Resource Access Management (RAM) ユーザーを使用する場合は、特権アカウントと標準アカウントを作成し、標準アカウントを RAM ユーザーに関連付ける必要があります。
ApsaraMQ for Kafka (Kafka) インスタンスを作成し、AnalyticDB for MySQL クラスターと同じ VPC にデプロイしていること。
Kafka トピックを作成し、メッセージを送信していること。詳細については、「Message Queue for Apache Kafka クイックスタート」をご参照ください。
注意事項
JSON 形式の Kafka データのみ同期できます。
Kafka トピック内のデータは、一定期間が経過すると自動的に削除されます。トピック内のデータが有効期限切れになり、データ同期タスクが失敗した場合、タスクを再起動しても有効期限切れのデータは読み取れません。これにより、データが失われる可能性があります。この問題を回避するには、トピックのデータライフサイクルを延長し、データ同期タスクが失敗した場合は速やかにテクニカルサポートにご連絡ください。
サンプル Kafka データが 8 KB を超える場合、Kafka API はデータを切り捨てます。これにより、サンプルデータの解析に失敗し、フィールドマッピング情報が自動的に生成されなくなります。
Kafka ソースのテーブルスキーマが変更されても、DDL 変更は自動的にトリガーされません。つまり、変更は AnalyticDB for MySQL に同期されません。
課金
AnalyticDB for MySQL クラスターの AnalyticDB コンピュートユニット (ACU) のエラスティックリソースの料金については、「Data Lakehouse Edition の課金項目」および「Enterprise Edition と Basic Edition の課金項目」をご参照ください。
操作手順
ステップ 1：データソースの作成
すでに Kafka データソースを追加している場合は、このステップをスキップして、新しい同期リンクを作成できます。
AnalyticDB for MySQL コンソールにログインします。コンソールの左上でリージョンを選択します。左側のナビゲーションウィンドウで、クラスターリスト をクリックします。管理するクラスターを見つけて、クラスター ID をクリックします。
左側のナビゲーションウィンドウで、[データインジェスト] > [データソース] を選択します。
右上隅にある データソースの新規作成 をクリックします。
データソースの新規作成 ページで、次のパラメーターを設定します：
パラメーター
説明
データソースのタイプ
データソースタイプとして [Kafka] を選択します。
データソース名
システムによって、データソースタイプと現在時刻に基づいて名前が生成されます。必要に応じて名前を変更できます。
データソースの説明
データソースの説明 (適用シナリオやビジネス上の制限など)。
デプロイモード
Alibaba Cloud インスタンスのみがサポートされています。
Kafka インスタンス
Kafka インスタンスの ID。
ApsaraMQ for Kafka コンソールにログインし、クラスターリスト ページでインスタンス ID を表示します。
Kafka Topic
Kafka で作成されたトピックの名前。
ApsaraMQ for Kafka コンソールにログインし、対象インスタンスの [トピック管理] ページでトピック名を表示します。
メッセージデータフォーマット
Kafka メッセージのデータ形式。JSON のみがサポートされています。
パラメーターを設定したら、作成 をクリックします。
ステップ 2：同期リンクの作成
左側のナビゲーションウィンドウで、Simple Log Service / Kafka データ同期 をクリックします。
左上隅にある 同期リンクの新規作成 をクリックし、Kafka データソース タブをクリックします。
同期リンクの新規作成 ページで、データソースと宛先の設定、ターゲットデータベースとターゲットテーブルの設定、同期設定 を設定します。
データソースと宛先の設定 セクションのパラメーターは次の表のとおりです。
パラメーター
説明
データリンク名
データリンクの名前。システムによって、データソースタイプと現在時刻に基づいて名前が生成されます。必要に応じて名前を変更できます。
データソース
既存の Kafka データソースを選択するか、新しいデータソースを作成します。
データソースフォーマット
JSON のみがサポートされています。
送信先ポートタイプ
選択： [データウェアハウス - ADB ストレージ]。
ADB アカウント
AnalyticDB for MySQL クラスターのデータベースアカウント。
ADB パスワード
AnalyticDB for MySQL クラスターのデータベースアカウントのパスワード。
ターゲットデータベースとターゲットテーブルの設定 セクションのパラメーターは次の表のとおりです。
パラメーター
説明
ライブラリ名
AnalyticDB for MySQL クラスターのデータベース名。
テーブル名
AnalyticDB for MySQL クラスターのテーブル名。
サンプルデータ
Kafka トピックから最新のデータが自動的に取得され、サンプルデータとして使用されます。
説明Kafka トピックのデータは JSON 形式である必要があります。他の形式のデータが存在する場合、データ同期中にエラーが発生します。
JSON 解析階層
JSON データを解析するネストされたレイヤーの数。有効な値：
0：解析なし。
1 (デフォルト)：1 つのレイヤーを解析します。
2：2 つのレイヤーを解析します。
3：3 つのレイヤーを解析します。
4：4 つのレイヤーを解析します。
JSON のネスト解析ポリシーの詳細については、「データ同期機能 (APS) を使用して Kafka データを同期する (推奨)」をご参照ください。
スキーマフィールドマッピング
JSON 解析後のサンプルデータのスキーマ情報を表示します。必要に応じて、宛先フィールド名とタイプを変更したり、フィールドを追加または削除したりできます。
[同期設定] セクションのパラメーターは次の表のとおりです。
パラメーター
説明
配信開始点
同期タスクが開始されると、選択した時点から Kafka データを消費します。任意の時点を選択でき、システムはその時点以降の Kafka の最初のデータレコードから消費を開始します。
ダーティデータ処理モード
データ同期中に、宛先テーブルのフィールドのデータ型がソース Kafka データの実際のデータ型と一致しない場合、同期は失敗します。たとえば、ソースデータが
abcで、宛先テーブルのフィールドタイプがintの場合、変換エラーが発生し、同期が異常になります。ダーティデータ処理モードは、次のいずれかの値に設定できます：
[同期を停止] (デフォルト)：データ同期が停止します。宛先テーブルのフィールドタイプを変更するか、ダーティデータ処理モードを変更してから、同期タスクを再起動する必要があります。
[NULL として処理]：ダーティデータを含むフィールドは、NULL 値として宛先テーブルに書き込まれます。
たとえば、Kafka データの行に 3 つのフィールド (col1、col2、col3) があり、col2 フィールドにダーティデータが含まれている場合、col2 フィールドのデータは NULL に変換されてテーブルに書き込まれます。col1 と col3 フィールドのデータは正常に書き込まれます。
ジョブ型リソースグループ
タスクを実行するジョブ固有のリソースグループを指定します。
増分同期に必要な ACU の数
タスクが実行されるジョブ固有のリソースグループの ACU 数。最小値は 2 ACU です。最大値は、ジョブ固有のリソースグループで利用可能な最大コンピューティングリソースです。データインジェストのパフォーマンスとタスクの安定性を向上させるために、ACU の数を増やすことを推奨します。
説明データ同期タスクを作成すると、ジョブ固有のリソースグループのエラスティックリソースが使用されます。データ同期タスクはリソースを長時間占有するため、システムはタスクが占有するリソースをリソースグループから差し引きます。たとえば、ジョブ固有のリソースグループの最大コンピューティングリソースが 48 ACU で、8 ACU を使用する同期タスクを作成した場合、このリソースグループで別の同期タスクに選択できる ACU の最大数は 40 です。
[ホワイトリストに追加]
データ同期のためのネットワーク接続を確立するには、Kafka vSwitch の CIDR ブロックを AnalyticDB for MySQL クラスターのホワイトリストに追加する必要があります。
パラメーターを設定したら、送信 をクリックします。
ステップ 3：データ同期タスクの開始
Simple Log Service / Kafka データ同期 ページで、作成したデータ同期タスクを見つけ、操作 列の スタート をクリックします。
左上隅にある クエリ をクリックします。タスクのステータスが [実行中] に変わります。これは、データ同期タスクが正常に開始されたことを示します。