Amazon Athena MSK コネクタ

Amazon MSK 用の Amazon Athena コネクタを使用すると、Amazon Athena で Apache Kafka トピックに対する SQL クエリを実行できるようになります。このコネクタを使用すると、Apache Kafka トピックをテーブルとして、メッセージを Athena の行として表示できます。追加情報については、「AWS ビッグデータブログ」の「Amazon Athena を使用して Amazon MSK でリアルタイムストリーミングデータを分析する」を参照してください。

前提条件

Athena コンソールまたは AWS Serverless Application Repository を使用して AWS アカウント にコネクタをデプロイします。詳細については、「データソースコネクタをデプロイする」または「AWS Serverless Application Repository を使用してデータソースコネクタをデプロイする」を参照してください。

制約事項

用語

クラスター間の互換性

MSK コネクタは、次のクラスタータイプで使用できます。

サポートされた認証方法

コネクタでは、次の認証方法がサポートされています。

対応する入力データ形式

コネクタは、次の入力データ形式をサポートします。

パラメータ

このセクションに示した Lambda 環境変数により、Athena MSK コネクタを設定できます。

サポートされるデータ型

次の表に、Kafka と Apache Arrow に対応するデータ型を示します。

パーティションと分割

Kafka トピックはパーティションに分割されています。各パーティションは順序付けられています。パーティション内の各メッセージには、オフセットと呼ばれるインクリメンタル ID があります。各 Kafka パーティションは、並列処理のためにさらに複数のスプリットに分割されます。データは、Kafka クラスターに設定された保持期間中、利用できます。

ベストプラクティス

ベストプラクティスとして、次の例のように、Athena にクエリを実行するときは、述語プッシュダウンを使用してください。

SELECT * 
FROM "msk_catalog_name"."glue_schema_registry_name"."glue_schema_name" 
WHERE integercol = 2147483647

SELECT * 
FROM "msk_catalog_name"."glue_schema_registry_name"."glue_schema_name" 
WHERE timestampcol >= TIMESTAMP '2018-03-25 07:30:58.878'

MSK コネクタのセットアップ

コネクタを使用する前に、Amazon MSK クラスターをセットアップし、AWS Glue スキーマレジストリを使用してスキーマを定義し、コネクタの認証を設定する必要があります。

AWS Glue スキーマレジストリを使用する際は、次の点に注意してください。

{
  "topicName": "datatypejson",
  "message": {
    "dataFormat": "json",
    "fields": [
      {
        "name": "intcol",
        "mapping": "intcol",
        "type": "INTEGER"
      },
      {
        "name": "varcharcol",
        "mapping": "varcharcol",
        "type": "VARCHAR"
      },
      {
        "name": "booleancol",
        "mapping": "booleancol",
        "type": "BOOLEAN"
      },
      {
        "name": "bigintcol",
        "mapping": "bigintcol",
        "type": "BIGINT"
      },
      {
        "name": "doublecol",
        "mapping": "doublecol",
        "type": "DOUBLE"
      },
      {
        "name": "smallintcol",
        "mapping": "smallintcol",
        "type": "SMALLINT"
      },
      {
        "name": "tinyintcol",
        "mapping": "tinyintcol",
        "type": "TINYINT"
      },
      {
        "name": "datecol",
        "mapping": "datecol",
        "type": "DATE",
        "formatHint": "yyyy-MM-dd"
      },
      {
        "name": "timestampcol",
        "mapping": "timestampcol",
        "type": "TIMESTAMP",
        "formatHint": "yyyy-MM-dd HH:mm:ss.SSS"
      }
    ]
  }
}

{
  "topicName": "datatypecsvbulk",
  "message": {
    "dataFormat": "csv",
    "fields": [
      {
        "name": "intcol",
        "type": "INTEGER",
        "mapping": "0"
      },
      {
        "name": "varcharcol",
        "type": "VARCHAR",
        "mapping": "1"
      },
      {
        "name": "booleancol",
        "type": "BOOLEAN",
        "mapping": "2"
      },
      {
        "name": "bigintcol",
        "type": "BIGINT",
        "mapping": "3"
      },
      {
        "name": "doublecol",
        "type": "DOUBLE",
        "mapping": "4"
      },
      {
        "name": "smallintcol",
        "type": "SMALLINT",
        "mapping": "5"
      },
      {
        "name": "tinyintcol",
        "type": "TINYINT",
        "mapping": "6"
      },
      {
        "name": "floatcol",
        "type": "DOUBLE",
        "mapping": "7"
      }
    ]
  }
}

Athena MSK コネクタの認証の設定

Amazon MSK クラスターへの認証には、IAM、SSL、SCRAM、スタンドアロン Kafka など、さまざまな方法を使用できます。

次の表は、コネクタの認証タイプと、それぞれのセキュリティプロトコルおよび SASL メカニズムを示しています。詳細については、「Amazon Managed Streaming for Apache Kafka Developer Guide」の「Authentication and authorization for Apache Kafka APIs」を参照してください。

SASL/IAM

クラスターが IAM 認証を使用する場合、クラスターをセットアップする際にユーザーの IAM ポリシーを設定する必要があります。詳細については、Amazon Managed Streaming for Apache Kafka デベロッパーガイドの「IAM access control」(IAM アクセスコントロール) を参照してください。

この認証タイプを使用するには、コネクタの auth_type Lambda 環境変数を SASL_SSL_AWS_MSK_IAM に設定します。

SSL

クラスターが SSL 認証されている場合、トラストストアとキーストアファイルを生成して Amazon S3 バケットにアップロードする必要があります。コネクタをデプロイする際に、この Amazon S3 リファレンスを提供する必要があります。キーストア、トラストストア、および SSL キーは AWS Secrets Manager に保存されます。コネクタをデプロイする際に、AWS のシークレットキーを提供します。

Secrets Manager でのシークレットの作成については、「AWS Secrets Manager シークレットを作成する」を参照してください。

この認証タイプを使用するには、以下の表の説明どおりに環境変数を設定します。

Secrets Manager でシークレットを作成した後、Secrets Manager コンソールでそのシークレットを確認できます。

SASL/SCRAM

クラスターが SCRAM 認証を使用している場合は、コネクタをデプロイするときに、クラスターに関連付けられている Secrets Manager キーを指定します。ユーザーの AWS 認証情報 (シークレットキーとアクセスキー) は、クラスターとの認証に使用されます。

下表のように環境変数を設定します。

次の画像は、Secrets Manager コンソールにある 2 つのキー/値のペア (1 つは username 用、もう 1 つは password 用) のシークレットの例を示しています。

ライセンス情報

このコネクタを使用することにより、pom.xml ファイル内のリストにある、サードパーティのコンポーネントが使用されることを承認し、 GitHub.com にある LICENSE.txt ファイルに記載されている、個別のサードパーティライセンスの使用条件に同意したとみなされます。

追加リソース

このコネクタに関するその他の情報については、GitHub.com で対応するサイトを参照してください。