ApsaraMQ for Kafka:MaxCompute sink コネクタの作成

RAM ロールの作成

Resource Access Management (RAM) ロールは、信頼できるサービスとして ApsaraMQ for Kafka を直接選択することをサポートしていません。 したがって、RAM ロールを作成するときは、サポートされている任意のサービスを信頼できるサービスとして選択します。 RAM ロールが作成された後、手動で信頼ポリシーを変更します。

1. RAM コンソールにログインします。

2. 左側のナビゲーションバーで、[ID] > [ロール] を選択します。

3. [ロール] ページで、[ロールの作成] をクリックします。

   次の図は、[ロール] ページの [ロールの作成] ボタンを示しています。

   

4. [ロールの作成] パネルで、次の操作を実行します。

5. 信頼できるエンティティタイプを [Alibaba Cloud サービス] に設定し、次へ をクリックします。

6. [ロールタイプ] セクションで、[通常のサービスロール] を選択します。 [ロール名] フィールドに `AliyunKafkaMaxComputeUser1` と入力します。 [信頼できるサービスの選択] リストから [MaxCompute] を選択し、[完了] をクリックします。

7. [ロール] ページで、[AliyunKafkaMaxComputeUser1] を見つけてクリックします。

8. [AliyunKafkaMaxComputeUser1] ページで、[信頼ポリシー] タブをクリックし、[信頼ポリシーの編集] をクリックします。

9. [信頼ポリシーの編集] パネルで、スクリプト内の odps を `alikafka` に置き換えてから、[OK] をクリックします。

   次のポリシーは、置換後の結果を示しています。

権限の追加

コネクタがメッセージを MaxCompute テーブルに同期できるようにするには、作成した RAM ロールに少なくとも次の権限を付与する必要があります。

上記の権限とその付与方法の詳細については、「MaxCompute の権限」をご参照ください。

以下の手順では、この Topic で作成された AliyunKafkaMaxComputeUser1 ロールに権限を付与する方法を説明します。 MaxCompute は RAM ロールをユーザーとして管理するため、このセクションのすべてのコマンドは RAM$<accountid>:role/aliyunkafkamaxcomputeuser1 オブジェクトを対象とします。

1. MaxCompute クライアントにログインします。

2. 次のコマンドを実行して、RAM ロールをユーザーとして追加します。

   add user `RAM$<accountid>:role/aliyunkafkamaxcomputeuser1`;

3. RAM ロールに MaxCompute へのアクセスに必要な最小権限を付与します。

4. 次のコマンドを実行して、RAM ロールにプロジェクトに対する権限を付与します。

grant CreateInstance on project connector_test to user `RAM$<accountid>:role/aliyunkafkamaxcomputeuser1`;

2. 次のコマンドを実行して、RAM ロールにテーブルに対する権限を付与します。

grant Describe, Alter, Update on table test_kafka to user `RAM$<accountid>:role/aliyunkafkamaxcomputeuser1`;

MaxCompute sink コネクタで必要なトピックの作成

ApsaraMQ for Kafka コンソールでは、MaxCompute sink コネクタで必要な 5 つのトピック (タスクオフセットトピック、タスク設定トピック、タスクステータストピック、デッドレターキュートピック、エラーデータトピック) を手動で作成できます。 必要なパーティション数とストレージエンジンは、次の表に示すようにトピックによって異なります。

トピックリソースを節約するために、1 つのトピックをデッドレターキュートピックとエラーデータトピックの両方として使用できます。 デッドレターキュートピックとエラーデータトピックの場合、ストレージエンジンはローカルストレージまたはクラウドストレージにすることができ、残りのプロパティは、このセクションのトピックプロパティテーブルで説明されている一般的なルールに従います。 各トピックの完全な説明については、「ソースサービスの設定タブのパラメーター」をご参照ください。

ApsaraMQ for Kafka コンソールにログインします。

ApsaraMQ for Kafka コンソールにログインします。

ApsaraMQ for Kafka コンソールにログインします。

ApsaraMQ for Kafka コンソールにログインします。

概要 ページの リソースの分布 セクションで、リージョンを選択します。

概要 ページの リソースの分布 セクションで、リージョンを選択します。

概要 ページの リソースの分布 セクションで、リージョンを選択します。

インスタンスリスト ページで、ターゲットインスタンスの名前をクリックします。

インスタンスリスト ページで、ターゲットインスタンスの名前をクリックします。

左側のナビゲーションウィンドウで、トピック管理 をクリックします。

各コネクタトピックのプロパティを前の表に記載されている値に設定します。 次の表に、すべてのトピックプロパティを示します。

Topic が作成されると、トピック管理 ページのリストに表示されます。これらの Topic は手動で作成されているため、コネクタを作成する際は リソースの作成方法 を 手動で作成します に設定し、作成した Topic の名前を入力してください。

MaxCompute sink コネクタで必要な グループの作成

ApsaraMQ for Kafka コンソールでは、MaxCompute sink コネクタのデータ同期タスクで使用される グループを手動で作成できます。 グループの名前は `connect-タスク名` である必要があります。ここで、`タスク名` はコネクタの名前です。 詳細については、「ソースサービスの設定タブのパラメーター」をご参照ください。

ApsaraMQ for Kafka コンソールにログインします。

ApsaraMQ for Kafka コンソールにログインします。

ApsaraMQ for Kafka コンソールにログインします。

ApsaraMQ for Kafka コンソールにログインします。

概要 ページの リソースの分布 セクションで、リージョンを選択します。

概要 ページの リソースの分布 セクションで、リージョンを選択します。

概要 ページの リソースの分布 セクションで、リージョンを選択します。

インスタンスリストページで、ターゲットインスタンスの名前をクリックします。

インスタンスリスト ページで、ターゲットインスタンスの名前をクリックします。

左側のナビゲーションウィンドウで、Group の管理 をクリックします。

グループが作成されると、Group の管理 ページのリストに表示されます。コネクタを作成する際に、Connector コンシューマーグループ フィールドにその名前を入力します。

MaxCompute sink コネクタの作成とデプロイ

ApsaraMQ for Kafka から MaxCompute にデータを同期する MaxCompute sink コネクタを作成してデプロイします。

ApsaraMQ for Kafka コンソールにログインします。

ApsaraMQ for Kafka コンソールにログインします。

ApsaraMQ for Kafka コンソールにログインします。

ApsaraMQ for Kafka コンソールにログインします。

概要ページのリソースの分布セクションで、リージョンを選択します。

概要 ページの リソースの分布 セクションで、リージョンを選択します。

概要 ページの リソースの分布 セクションで、リージョンを選択します。

左側のナビゲーションウィンドウで、Connector タスクリスト をクリックします。

左側のナビゲーションウィンドウで、Connector タスクリスト をクリックします。

Connector タスクリスト ページで、インスタンスの選択 ドロップダウンリストからコネクタが属するインスタンスを選択し、Connector の作成をクリックします。

1. Connector の作成 構成ウィザードページで、以下の操作を実行します。

1. 基本情報の設定 タブで、ビジネス要件に基づいて次のパラメーターを設定し、次へ をクリックします。

   パラメーター	説明	例
   名前	コネクタの名前。 命名規則： - 名前には数字、小文字、ハイフン (-) を含めることができますが、ハイフン (-) で始めることはできません。 名前の長さは 1～48 文字である必要があります。 - 名前は、同じ ApsaraMQ for Kafka インスタンス内で一意である必要があります。 コネクタのデータ同期タスクは、`connect-タスク名` という名前の グループを使用する必要があります。 グループを手動で作成しない場合、システムが自動的に作成します。	kafka-maxcompute-sink
   インスタンス	デフォルトでは、インスタンス名とインスタンス ID が指定されます。	demo alikafka_post-cn-st21p8vj****

2. [ソースサービスの設定] タブで、[データソース] を [ApsaraMQ for Kafka] に設定し、次のパラメーターを設定してから 次へ をクリックします。

   説明

   すでに Topic とコンシューマーグループを作成している場合は、手動でのリソース作成を選択し、既存リソースの情報を入力してください。それ以外の場合は、自動でのリソース作成を選択してください。

   すでに Topic と コンシューマーグループを作成している場合は、手動リソース作成を選択し、既存リソースの情報を入力してください。そうでない場合は、自動リソース作成を選択してください。

   すでに Topic と コンシューマーグループ を作成済みの場合は、手動でのリソース作成を選択し、既存リソースの情報を入力してください。それ以外の場合は、自動でのリソース作成を選択してください。

   すでに Topic と コンシューマーグループ を作成している場合は、手動リソース作成を選択して、既存リソースの情報を入力します。それ以外の場合は、自動リソース作成を選択します。

   すでに Topic と コンシューマーグループを作成している場合は、手動リソース作成を選択し、既存リソースの情報を入力してください。それ以外の場合は、自動リソース作成を選択してください。

   パラメーター	説明	例
   データソース Topic	同期したいデータのソーストピック。	maxcompute-test-input
   コンシューマースレッドの同時発生数	ソーストピックの同時コンシューマースレッド数。 デフォルト値：6。 有効な値： - 1 - 2 - 3 - 6 - 12	6
   消費の開始位置	メッセージが消費されるオフセット。 有効な値： - 一番古いオフセット：最小オフセットから消費を開始します。 - 一番新しいオフセット：最大オフセットから消費を開始します。	一番古いオフセット
   VPC ID	データ同期タスクが実行される VPC。 このパラメーターを表示するには、実行環境の設定 をクリックします。 デフォルトでは、ApsaraMQ for Kafka インスタンスがデプロイされている VPC が使用されます。 このパラメーターを指定する必要はありません。	vpc-bp1xpdnd3l***
   VSwitch ID	データ同期タスクが実行される vSwitch。 このパラメーターを表示するには、実行環境の設定 をクリックします。 vSwitch は、ApsaraMQ for Kafka インスタンスと同じ VPC にある必要があります。 デフォルトでは、ApsaraMQ for Kafka インスタンスをデプロイしたときに指定した vSwitch が使用されます。	vsw-bp1d2jgg81***
   失敗の処理	メッセージの送信に失敗した後、エラーが発生したトピックのパーティションへのサブスクリプションを続行するかどうかを指定します。 このパラメーターを表示するには、実行環境の設定 をクリックします。 有効な値： - サブスクリプションの継続：エラーが発生したトピックのパーティションへのサブスクリプションを続行し、エラーログを出力します。 - サブスクリプションの停止：エラーが発生したトピックのパーティションへのサブスクリプションを停止し、エラーログを出力します > 注： - ログの表示方法の詳細については、「コネクタ操作」をご参照ください。 - エラーコードに基づいた解決策の検索方法の詳細については、「エラーコード」をご参照ください。 - エラーが発生したトピックのパーティションへのサブスクリプションを再開するには、して テクニカルサポートにお問い合わせください。	サブスクリプションの継続
   リソースの作成方法	コネクタで必要なトピックと グループを作成する方法。 このパラメーターを表示するには、実行環境の設定 をクリックします。 有効な値： - 自動作成：システムが必要なトピックと グループを自動的に作成します。 事前に作成していない場合は、この値を選択します。 - 手動で作成します：事前にトピックと グループを作成した場合は、この値を選択し、次のフィールドにそれらの名前を入力します。	自動作成
   Connector コンシューマーグループ	コネクタのデータ同期タスクで使用される グループ。 このパラメーターを表示するには、実行環境の設定 をクリックします。 グループの名前は `connect-タスク名` である必要があります。	connect-kafka-maxcompute-sink
   タスクサイトの Topic	コンシューマーオフセットを保存するトピック。 このパラメーターを表示するには、実行環境の設定 をクリックします。 - トピック名：名前は connect-offset で始まることを推奨します。 - パーティション数：トピックのパーティション数は 1 より大きい必要があります。 - ストレージエンジン：トピックのストレージエンジンはローカルストレージである必要があります。 - cleanup.policy：トピックのログクリーンアップポリシーは compact である必要があります。	connect-offset-kafka-maxcompute-sink
   タスク設定の Topic	タスク設定を保存するトピック。 このパラメーターを表示するには、実行環境の設定 をクリックします。 - トピック名：名前は connect-config で始まることを推奨します。 - パーティション数：トピックのパーティション数は 1 である必要があります。 - ストレージエンジン：トピックのストレージエンジンはローカルストレージである必要があります。 - cleanup.policy：トピックのログクリーンアップポリシーは compact である必要があります。	connect-config-kafka-maxcompute-sink
   タスクステータスの Topic	タスクステータスを保存するトピック。 このパラメーターを表示するには、実行環境の設定 をクリックします。 - トピック名：名前は connect-status で始まることを推奨します。 - パーティション数：トピックのパーティション数を 6 に設定することを推奨します。 - ストレージエンジン：トピックのストレージエンジンはローカルストレージである必要があります。 - cleanup.policy：トピックのログクリーンアップポリシーは compact である必要があります。	connect-status-kafka-maxcompute-sink
   デッドレターキューの Topic	Connect フレームワークの異常データを保存するトピック。 このパラメーターを表示するには、実行環境の設定 をクリックします。 トピックリソースを節約するために、このトピックをエラーデータトピックとして使用できます。 - トピック名：名前は connect-error で始まることを推奨します。 - パーティション数：トピックのパーティション数を 6 に設定することを推奨します。 - ストレージエンジン：トピックのストレージエンジンはローカルストレージまたはクラウドストレージにすることができます。	connect-error-kafka-maxcompute-sink
   例外データ Topic	sink の異常データを保存するトピック。 このパラメーターを表示するには、実行環境の設定 をクリックします。 トピックリソースを節約するために、このトピックをデッドレターキュートピックとして使用できます。 - トピック名：名前は connect-error で始まることを推奨します。 - パーティション数：トピックのパーティション数を 6 に設定することを推奨します。 - ストレージエンジン：トピックのストレージエンジンはローカルストレージまたはクラウドストレージにすることができます。	connect-error-kafka-maxcompute-sink

3. ターゲットサービスの設定 タブで、宛先サービスを MaxCompute に設定し、以下のパラメーターを設定した後、作成 をクリックします。

コネクタが作成されると、Connector タスクリスト ページで表示できます。

2. コネクタが作成された後、Connector タスクリスト ページに移動し、作成したコネクタを見つけて、操作 列の デプロイ をクリックします。

テストメッセージの送信

MaxCompute sink コネクタをデプロイした後、ApsaraMQ for Kafka のソーストピックにメッセージを送信して、データが MaxCompute に同期できるかどうかをテストできます。

Connector タスクリスト ページで、目的のコネクタを探し、操作 列の テスト をクリックします。

Connector タスクリスト ページで、目的のコネクタを見つけ、操作 列の テスト をクリックします。

Connector タスクリスト ページで、目的のコネクタを見つけ、操作 列の テスト をクリックします。

Connector タスクリスト ページで、対象のコネクタを見つけ、操作 列の テスト をクリックします。

1. メッセージの送信 パネルで、テストメッセージを送信します。

送信方法 を コンソール に設定します。

• メッセージキー テキストボックスに、メッセージのキー (たとえば demo) を入力します。

• メッセージの内容 フィールドに、テストメッセージの内容 (例: {"key": "test"}) を入力します。

• メッセージを特定のパーティションに送信するかどうかを指定するには、指定されたパーティションに送信 を設定します。

• はい をクリックし、パーティション ID フィールドにパーティション ID (たとえば 0) を入力します。 パーティション ID のクエリ方法の詳細については、「パーティションステータスの表示」をご参照ください。

• いいえ をクリックして、パーティションを指定せずにメッセージを送信します。

  送信方法 を Docker に設定し、Docker コンテナーを実行してサンプルメッセージを生成する セクションで Docker コマンドを実行してメッセージを送信します。

送信方法 を SDK に設定します。 業務要件に応じて、使用するプログラミング言語またはフレームワークのソフトウェア開発キット (SDK) と接続方法を選択し、その SDK を使用してメッセージを送信します。

テーブルデータの表示

ApsaraMQ for Kafka のソーストピックにメッセージを送信した後、MaxCompute クライアントでテーブルデータを表示して、メッセージが受信されたかどうかを確認します。

以下の手順では、このトピックで `test_kafka` に書き込まれたデータを表示する方法を示します。 コネクタは、[パーティション] パラメーターに設定した粒度で、メッセージが書き込まれた時間に対応するパーティションに各メッセージを書き込みます。

1. MaxCompute クライアントにログインします。

2. 次のコマンドを実行して、テーブルのデータパーティションを表示します。

   show partitions test_kafka;

   次の結果が返されます。

   pt=11-17-2020 15
   
   OK

3. 次のコマンドを実行して、前の手順で返されたパーティションのデータを表示します。

select * from test_kafka where pt ="11-17-2020 15";

次の結果が返されます。

+----------------------+------------+------------+-----+-------+---------------+
| topic                | partition  | offset     | key | value | pt            |
+----------------------+------------+------------+-----+-------+---------------+
| maxcompute-test-input| 0          | 0          | 1   | 1     | 11-17-2020 15 |
+----------------------+------------+------------+-----+-------+---------------+

クエリで行が返されなかった場合は、show partitions が返したパーティションをクエリしたことを確認し、ソース Topic をサブスクライブする グループ の消費の詳細を確認し、コネクタのエラーデータ Topic とデッドレターキュー Topic に書き込みに失敗したメッセージがないかを確認してください。