メインコンテンツまでスキップ
に最終更新

AWS MSK

この機能は Doris 4.1.0 以降でサポートされます。

Amazon Managed Streaming for Apache Kafka（AWS MSK）は、AWS が提供するフルマネージドの Apache Kafka サービスです。Doris は Routine Load を通じて AWS MSK からのリアルタイムデータ取り込みをサポートし、AWS MSK の IAM 認証を利用できます。CSV 形式と JSON 形式に対応し、データの欠落や重複を防ぐ Exactly-Once セマンティクスを提供します。詳しくは Routine Load を参照してください。

前提条件​

Routine Load を使用して AWS MSK からデータを取り込む前に、次の条件を満たしていることを確認してください。

  1. AWS MSK クラスターが作成され、IAM 認証が有効になっている。
  2. MSK クラスターへのアクセスを許可する適切な AWS IAM 権限が設定されている。
  3. Doris クラスターから AWS MSK の Bootstrap Servers にアクセスできる。

認証パラメータ​

次の表に、AWS MSK からデータを取り込む際に設定する認証関連のパラメータを示します。

パラメータ説明例
aws.regionAWS リージョン"us-east-1"
aws.access_keyAWS アクセスキー ID-
aws.secret_keyAWS シークレットアクセスキー-
aws.role_arnアカウントをまたぐアクセスの認証情報に使用するロール"arn:aws:iam::123456789012:role/MyRole"
aws.profile_name~/.aws/credentials に設定された AWS プロファイル名-
aws.credentials_providerAWS SDK の標準認証情報プロバイダー。複数のプロバイダー種別に対応"INSTANCE_PROFILE"
aws.external_idAssumeRole の「呼び出しコンテキスト識別子」として使用-
property.security.protocolIAM 認証の制約により、SASL_SSL に固定"SASL_SSL"
property.sasl.mechanismlibrdkafka ライブラリの制約により、OAUTHBEARER に固定"OAUTHBEARER"

aws.credentials_provider に指定できる値​

値説明
DEFAULTデフォルトのプロバイダーチェーンを使用
ENV環境変数から認証情報を読み取る
INSTANCE_PROFILEEC2 インスタンスプロファイルの認証情報を使用

認証設定の例​

Doris は次の IAM 認証方法をサポートします。実際のデプロイ環境に応じて一つ選択してください。

認証方法適用シナリオ
AK/SK を直接使用長期有効なアクセスキーとシークレットキーをすでに持っている場合
IAM ロール（Assume Role）アカウントをまたいでアクセスする場合、または一時的な認証情報を使用したい場合
aws.credentials_provider で認証情報の取得元を指定EC2 インスタンスプロファイルを使用する場合など、AK/SK を明示的に入力したくない場合

方法 1: アクセスキーとシークレットキー（AK/SK）を直接使用する​

CREATE ROUTINE LOAD IAM_Test ON t
COLUMNS TERMINATED BY ",",
COLUMNS(a,b)
FROM KAFKA(
    "kafka_broker_list" = "your_msk_broker_list",
    "kafka_topic" = "your_kafka_topic",

    "aws.region" = "us-west-1",
    "aws.access_key" = "<your-ak>",
    "aws.secret_key" = "<your-sk>",

    "property.kafka_default_offsets" = "OFFSET_BEGINNING",
    "property.security.protocol" = "SASL_SSL",
    "property.sasl.mechanism" = "OAUTHBEARER"
);

方法 2: IAM ロール（Assume Role）を使用する​

aws.role_arn を設定した場合、aws.credentials_provider は STS の AssumeRole 呼び出しに使用する元の認証情報プロバイダーを指定します。

例 2.1: EC2 インスタンスプロファイルを STS の認証情報の取得元として使用する​

CREATE ROUTINE LOAD IAM_Test ON t
COLUMNS TERMINATED BY ",",
COLUMNS(a,b)
FROM KAFKA(
    "kafka_broker_list" = "your_msk_broker_list",
    "kafka_topic" = "your_kafka_topic",

    "aws.region" = "us-west-1",
    "aws.role_arn" = "arn:aws:iam::123456789012:role/demo-role",
    "aws.credentials_provider" = "INSTANCE_PROFILE",

    "property.kafka_default_offsets" = "OFFSET_BEGINNING",
    "property.security.protocol" = "SASL_SSL",
    "property.sasl.mechanism" = "OAUTHBEARER"
);

例 2.2: 環境変数から AK/SK を読み取り、STS の認証情報の取得元として使用する​

CREATE ROUTINE LOAD IAM_Test ON t
COLUMNS TERMINATED BY ",",
COLUMNS(a,b)
FROM KAFKA(
    "kafka_broker_list" = "your_msk_broker_list",
    "kafka_topic" = "your_kafka_topic",

    "aws.region" = "us-west-1",
    "aws.role_arn" = "arn:aws:iam::123456789012:role/demo-role",
    "aws.credentials_provider" = "ENV",

    "property.kafka_default_offsets" = "OFFSET_BEGINNING",
    "property.security.protocol" = "SASL_SSL",
    "property.sasl.mechanism" = "OAUTHBEARER"
);

例 2.3: デフォルトのプロバイダーチェーンを STS の認証情報の取得元として使用する​

CREATE ROUTINE LOAD IAM_Test ON t
COLUMNS TERMINATED BY ",",
COLUMNS(a,b)
FROM KAFKA(
    "kafka_broker_list" = "your_msk_broker_list",
    "kafka_topic" = "your_kafka_topic",

    "aws.region" = "us-west-1",
    "aws.role_arn" = "arn:aws:iam::123456789012:role/demo-role",
    "aws.credentials_provider" = "DEFAULT",

    "property.kafka_default_offsets" = "OFFSET_BEGINNING",
    "property.security.protocol" = "SASL_SSL",
    "property.sasl.mechanism" = "OAUTHBEARER"
);

方法 3: aws.credentials_provider で認証情報の取得元を指定する​

EC2 インスタンスプロファイルを使用する場合など、AK/SK を明示的に指定しないシナリオに適しています。

CREATE ROUTINE LOAD IAM_Test ON t
COLUMNS TERMINATED BY ",",
COLUMNS(a,b)
FROM KAFKA(
    "kafka_broker_list" = "your_msk_broker_list",
    "kafka_topic" = "your_kafka_topic",

    "aws.region" = "us-west-1",
    "aws.credentials_provider" = "INSTANCE_PROFILE",

    "property.kafka_default_offsets" = "OFFSET_BEGINNING",
    "property.security.protocol" = "SASL_SSL",
    "property.sasl.mechanism" = "OAUTHBEARER"
);

認証情報の解決順序​

複数の認証情報パラメータを同時に設定した場合、次の優先順位で適用されます。

  1. aws.access_key と aws.secret_key の両方が設定されている場合、AK/SK が最優先で使用されます。
  2. AK/SK が設定されておらず、aws.role_arn が設定されている場合、IAM ロールが使用されます。この場合、aws.credentials_provider で STS の認証情報の取得元を選択します。
  3. AK/SK と aws.role_arn のいずれも設定されていない場合、aws.credentials_provider が AWS クライアントで使用するプロバイダーを直接決定します。

パブリックネットワークからのアクセスに関するトラブルシューティング​

パブリックネットワーク環境から AWS MSK にアクセスする場合、データ取り込み中に AWS 認証の問題が発生したら、次の手順で原因を調べてください。

手順 1: MSK クラスターでパブリックアクセスが有効になっていることを確認する​

AWS MSK コンソールでアクセス先のクラスターを選択し、Properties の Networking settings にある Edit public access を開きます。パブリックアクセスのオプションが有効になっていることを確認してください。

手順 2: サブネットがパブリックであることを確認する​

クラスターに関連付けられたサブネットはパブリックである必要があります。AWS VPC コンソールで、サブネットのルートテーブルに 0.0.0.0/0 : igw-xxxx のエントリが含まれていることを確認してください。

手順 3: 正しいパブリック Bootstrap エンドポイントを使用する​

AWS MSK コンソールでアクセス先のクラスターを選択し、View client information をクリックします。Routine Load の作成時に指定する kafka_broker_list プロパティには、プライベートエンドポイントではなくパブリックエンドポイントを設定してください。

手順 4: セキュリティグループのインバウンドルールとアウトバウンドルールが正しく設定されていることを確認する​

MSK に設定されたセキュリティグループのインバウンドルールを確認し、ポート 9198 に対して適切な送信元 IP が設定されていることを確認してください。

注: IAM アクセス制御を使用してブローカーと通信する場合は、ポート 9198 経由のアクセスを許可する必要があります。

参考資料​

詳しくは、関連する AWS ドキュメントを参照してください。

よくある質問​

Q1: property.security.protocol と property.sasl.mechanism の値を固定する必要があるのはなぜですか？​

AWS MSK の IAM 認証方式と、基盤となる librdkafka ライブラリの制約により、これらのパラメータはそれぞれ SASL_SSL と OAUTHBEARER に固定する必要があります。それ以外の値では、IAM 認証のハンドシェイクを正常に完了できません。

Q2: AK/SK と aws.role_arn を同時に設定した場合、どちらの認証情報が使用されますか？​

AK/SK が優先して使用されます。詳しくは認証情報の解決順序を参照してください。

Q3: パブリックネットワークから MSK にアクセスすると認証が必ず失敗します。どう調べればよいですか？​

パブリックネットワークからのアクセスに関するトラブルシューティングの四つの手順に従い、パブリックアクセスが有効か、サブネットがパブリックか、パブリック Bootstrap エンドポイントを使用しているか、セキュリティグループでポート 9198 が許可されているかを順に確認してください。

Q4: EC2 で Doris を実行する場合、AK/SK を設定せずに済ませるにはどうすればよいですか？​

MSK へのアクセス権限を持つ IAM ロールを EC2 インスタンスに関連付け、aws.credentials_provider を INSTANCE_PROFILE に設定します。方法 3を参照してください。

Q5: Doris は Routine Load を使った MSK からのデータ取り込みで、データの欠落や重複がないことを保証しますか？​

はい。Routine Load は Exactly-Once セマンティクスを提供し、データの欠落や重複を防ぎます。

Report issue
On This Page