Amazon MSK と ClickHouse の連携

注: ビデオで示しているポリシーは権限が広く、クイックスタート専用です。最小権限の IAM ガイダンスについては以下を参照してください。

前提条件

以下を前提としています。

ClickHouse Connector Sinkについて理解していること
Amazon MSK および MSK Connectors について理解していること。Amazon MSK の Getting Started ガイドと MSK Connect ガイドを参照することを推奨します。

Amazon MSK向け ClickHouse 公式 Kafkaコネクタ

接続情報を確認する

HTTP(S) で ClickHouse に接続するには、次の情報が必要です。

Parameter(s)	Description
`HOST` and `PORT`	通常、TLS を使用する場合のポートは 8443、TLS を使用しない場合は 8123 です。
`DATABASE NAME`	デフォルトでは `default` という名前のデータベースがあります。接続先のデータベース名を使用してください。
`USERNAME` and `PASSWORD`	デフォルトのユーザー名は `default` です。用途に応じたユーザー名を使用してください。

ClickHouse Cloud サービスの詳細は、ClickHouse Cloud コンソールで確認できます。サービスを選択し、Connect をクリックします。

HTTPS を選択します。接続情報は curl コマンドの例として表示されます。

セルフマネージド ClickHouse を使用している場合、接続情報は ClickHouse 管理者によって設定されます。

手順

ClickHouse Connector Sink をあらかじめ確認しておきます。
MSK インスタンスを作成します。
IAM ロールを作成して割り当てます。
ClickHouse Connect Sink の Release page から jar ファイルをダウンロードします。
ダウンロードした jar ファイルを、Amazon MSK コンソールの Custom plugin page からインストールします。
コネクタがパブリックな ClickHouse インスタンスと通信する場合は、インターネットアクセスを有効にします。
config で、トピック名、ClickHouse インスタンスのホスト名、パスワードを指定します。

connector.class=com.clickhouse.kafka.connect.ClickHouseSinkConnector
tasks.max=1
topics=<topic_name>
ssl=true
security.protocol=SSL
hostname=<hostname>
database=<database_name>
password=<password>
ssl.truststore.location=/tmp/kafka.client.truststore.jks
port=8443
value.converter.schemas.enable=false
value.converter=org.apache.kafka.connect.json.JsonConverter
exactlyOnce=true
username=default
schemas.enable=false

推奨される IAM 権限 (最小権限)

構成に必要な最小限の権限のみを使用してください。まずは以下の基本構成を出発点とし、必要に応じてオプションのサービスだけを追加してください。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "MSKClusterAccess",
      "Effect": "Allow",
      "Action": [
        "kafka:DescribeCluster",
        "kafka:GetBootstrapBrokers",
        "kafka:DescribeClusterV2",
        "kafka:ListClusters",
        "kafka:ListClustersV2"
      ],
      "Resource": "*"
    },
    {
      "Sid": "KafkaAuthorization",
      "Effect": "Allow",
      "Action": [
        "kafka-cluster:Connect",
        "kafka-cluster:DescribeCluster",
        "kafka-cluster:DescribeGroup",
        "kafka-cluster:DescribeTopic",
        "kafka-cluster:ReadData"
      ],
      "Resource": "*"
    },
    {
      "Sid": "OptionalGlueSchemaRegistry",
      "Effect": "Allow",
      "Action": [
        "glue:GetSchema*",
        "glue:ListSchemas",
        "glue:ListSchemaVersions"
      ],
      "Resource": "*"
    },
    {
      "Sid": "OptionalSecretsManager",
      "Effect": "Allow",
      "Action": [
        "secretsmanager:GetSecretValue"
      ],
      "Resource": [
        "arn:aws:secretsmanager:<region>:<account-id>:secret:<your-secret-name>*"
      ]
    },
    {
      "Sid": "OptionalS3Read",
      "Effect": "Allow",
      "Action": [
        "s3:GetObject"
      ],
      "Resource": "arn:aws:s3:::<your-bucket>/<optional-prefix>/*"
    }
  ]
}

AWS Glue Schema Registry を使用する場合にのみ、Glue ブロックを使用してください。
Secrets Manager から認証情報やトラストストアを取得する場合にのみ、Secrets Manager ブロックを使用してください。ARN は必要な範囲に限定してください。
S3 からアーティファクト (例: トラストストア) を読み込む場合にのみ、S3 ブロックを使用してください。バケット/プレフィックスの範囲に限定してください。

関連項目: Kafka ベストプラクティス – IAM.

パフォーマンスチューニング

パフォーマンスを向上させる方法の1つは、次の内容を ワーカー 設定に追加して、バッチサイズと Kafka から取得するレコード数を調整することです。

consumer.max.poll.records=[NUMBER OF RECORDS]
consumer.max.partition.fetch.bytes=[NUMBER OF RECORDS * RECORD SIZE IN BYTES]

使用する具体的な値は、必要なレコード数やレコードサイズによって異なります。たとえば、デフォルト値は次のとおりです。

consumer.max.poll.records=500
consumer.max.partition.fetch.bytes=1048576

実装面やその他の考慮事項を含む詳細については、公式のKafka および Amazon MSK のドキュメントを参照してください。

MSK Connect のネットワークに関する注意事項

MSK Connect から ClickHouse に接続するには、MSK クラスターを、インターネットアクセス用の Private NAT に接続されたプライベートサブネット内に配置することを推奨します。設定手順は以下のとおりです。なお、パブリックサブネットもサポートされていますが、ENI に Elastic IP アドレスを継続的に割り当てる必要があるため、推奨されません。詳しくは AWS のドキュメントを参照してください。

プライベートサブネットを作成する: VPC 内に新しいサブネットを作成し、プライベートサブネットとして設定します。このサブネットはインターネットに直接アクセスできないようにする必要があります。
NAT ゲートウェイを作成する: VPC のパブリックサブネット内に NAT ゲートウェイを作成します。NAT ゲートウェイにより、プライベートサブネット内のインスタンスはインターネットや他の AWS サービスに接続できますが、インターネット側からそれらのインスタンスへの接続開始は防止されます。
ルートテーブルを更新する: インターネット宛てのトラフィックを NAT ゲートウェイに転送するルートを追加します
セキュリティグループとネットワーク ACL の設定を確認する: 必要なトラフィックを許可するように、セキュリティグループとネットワーク ACL (アクセス制御リスト) を設定します。
1. MSK Connect ワーカーの ENI から、TLS ポート (通常は 9094) の MSK ブローカーへの通信。
2. MSK Connect ワーカーの ENI から、ClickHouse エンドポイント (9440 (native TLS) または 8443 (HTTPS) ) への通信。
3. MSK Connect ワーカーの SG からブローカーの SG へのインバウンドを許可する。
4. セルフホストの ClickHouse の場合は、サーバーで設定したポート (HTTP のデフォルトは 8123) を開放します。
セキュリティグループを MSK に関連付ける: これらのセキュリティグループが MSK クラスターおよび MSK Connect ワーカーに関連付けられていることを確認します。
ClickHouse Cloud への接続性:
1. パブリックエンドポイント + IP 許可リスト: プライベートサブネットからの NAT 経由のアウトバウンド通信が必要です。
2. 利用可能な場合はプライベート接続 (例: VPC peering/PrivateLink/VPN) 。VPC の DNS ホスト名と名前解決が有効になっており、DNS でプライベートエンドポイントを解決できることを確認してください。
接続性を検証する (クイックチェックリスト) :
1. コネクタ環境から、MSK の bootstrap DNS を名前解決し、ブローカーのポートに TLS で接続する。
2. ClickHouse の 9440 番ポート (HTTPS の場合は 8443) への TLS 接続を確立する。
3. AWS サービス (Glue/Secrets Manager) を使用する場合は、それらのエンドポイントへのアウトバウンド通信を許可する。