Kafka → Amazon Redshift 数据摄取 · 三方案对比与部署手册

aws kafka list-clusters-v2 --region us-east-1
aws kafka get-bootstrap-brokers --region us-east-1 --cluster-arn <ClusterArn>
# 输出示例(已脱敏):
# "BootstrapBrokerStringSaslIam": "boot-xxxxxxxx.c2.kafka-serverless.us-east-1.amazonaws.com:9098"

aws redshift-serverless list-workgroups --region us-east-1 \
  --query 'workgroups[].{name:workgroupName，evr:enhancedVpcRouting}'

# 从 EC2 实测 9098 连通(SSM 下发)
aws ssm send-command --region us-east-1 --instance-ids i-xxxxxxxx \
  --document-name AWS-RunShellScript \
  --parameters 'commands=["timeout 8 bash -c \"cat < /dev/null > /dev/tcp/<bootstrap 主机>/9098\" && echo TCP_OK || echo TCP_FAIL"]'

{
  "Version": "2012-10-17"，
  "Statement": [{
    "Effect": "Allow"，
    "Principal": { "Service": "redshift.amazonaws.com" }，
    "Action": "sts:AssumeRole"
  }]
}

{
  "Version": "2012-10-17"，
  "Statement": [
    { "Sid":"Connect"，"Effect":"Allow"，
      "Action":["kafka-cluster:Connect"，"kafka-cluster:DescribeCluster"]，
      "Resource":"arn:aws:kafka:us-east-1:<ACCOUNT_ID>:cluster/my-msk/*" }，
    { "Sid":"ReadTopic"，"Effect":"Allow"，
      "Action":["kafka-cluster:ReadData"，"kafka-cluster:DescribeTopic"]，
      "Resource":"arn:aws:kafka:us-east-1:<ACCOUNT_ID>:topic/my-msk/*" }，
    { "Sid":"Group"，"Effect":"Allow"，
      "Action":["kafka-cluster:AlterGroup"，"kafka-cluster:DescribeGroup"]，
      "Resource":"arn:aws:kafka:us-east-1:<ACCOUNT_ID>:group/my-msk/*" }
  ]
}

aws iam create-role --role-name RedshiftMSKStreamingRole \
  --assume-role-policy-document file://redshift-trust-policy.json
aws iam create-policy --policy-name RedshiftMSKStreamingPolicy \
  --policy-document file://redshift-msk-policy.json
aws iam attach-role-policy --role-name RedshiftMSKStreamingRole \
  --policy-arn arn:aws:iam::<ACCOUNT_ID>:policy/RedshiftMSKStreamingPolicy

# 挂到 namespace(保留原有默认角色)
aws redshift-serverless update-namespace --region us-east-1 \
  --namespace-name <namespace> \
  --default-iam-role-arn <原有默认角色 ARN> \
  --iam-roles <原有默认角色 ARN> arn:aws:iam::<ACCOUNT_ID>:role/RedshiftMSKStreamingRole

{
  "Version":"2012-10-17"，
  "Statement":[
    {"Sid":"Cluster"，"Effect":"Allow"，
     "Action":["kafka-cluster:Connect"，"kafka-cluster:AlterCluster"，"kafka-cluster:DescribeCluster"]，
     "Resource":"arn:aws:kafka:us-east-1:<ACCOUNT_ID>:cluster/my-msk/*"}，
    {"Sid":"Topic"，"Effect":"Allow"，
     "Action":["kafka-cluster:*Topic*"，"kafka-cluster:WriteData"，"kafka-cluster:ReadData"]，
     "Resource":"arn:aws:kafka:us-east-1:<ACCOUNT_ID>:topic/my-msk/*"}，
    {"Sid":"Group"，"Effect":"Allow"，
     "Action":["kafka-cluster:AlterGroup"，"kafka-cluster:DescribeGroup"]，
     "Resource":"arn:aws:kafka:us-east-1:<ACCOUNT_ID>:group/my-msk/*"}
  ]
}

aws iam create-policy --policy-name EC2MSKProducerPolicy \
  --policy-document file://ec2-msk-producer-policy.json
aws iam attach-role-policy --role-name <EC2 实例角色> \
  --policy-arn arn:aws:iam::<ACCOUNT_ID>:policy/EC2MSKProducerPolicy

aws ssm send-command --region us-east-1 --instance-ids i-xxxxxxxx \
  --document-name AWS-RunShellScript \
  --parameters 'commands=[
    "dnf install -y python3-pip"，
    "python3 -m pip install \"kafka-python==2.0.2\" aws-msk-iam-sasl-signer-python"，
    "python3 -c \"import kafka; print(kafka.__version__)\""
  ]'

import json， time， random， datetime
from kafka import KafkaProducer
from kafka.admin import KafkaAdminClient， NewTopic
from kafka.errors import TopicAlreadyExistsError， UnknownTopicOrPartitionError
from aws_msk_iam_sasl_signer import MSKAuthTokenProvider

BOOTSTRAP = 'boot-xxxxxxxx.c2.kafka-serverless.us-east-1.amazonaws.com:9098'
REGION = 'us-east-1'
TOPIC = 'orders-topic'

class TP:
    def token(self):
        t， _ = MSKAuthTokenProvider.generate_auth_token(REGION); return t

common = dict(bootstrap_servers=BOOTSTRAP， security_protocol='SASL_SSL'，
              sasl_mechanism='OAUTHBEARER'， sasl_oauth_token_provider=TP())

# 清空:删除 topic 再重建(offset 归零)
admin = KafkaAdminClient(**common， client_id='admin1')
try: admin.delete_topics([TOPIC])
except UnknownTopicOrPartitionError: pass
for _ in range(30):
    time.sleep(3)
    if TOPIC not in admin.list_topics(): break
for _ in range(20):
    try:
        admin.create_topics([NewTopic(name=TOPIC， num_partitions=1， replication_factor=3)]); break
    except TopicAlreadyExistsError: time.sleep(3)
admin.close()

surnames=['张'，'王'，'李'，'赵'，'刘'，'陈'，'杨'，'黄'，'周'，'吴']; names=['伟'，'芳'，'娜'，'强'，'磊'，'军'，'洋'，'勇'，'艳'，'杰']
provinces=[('广东省'，'深圳市'，'南山区')，('北京市'，'北京市'，'朝阳区')，('浙江省'，'杭州市'，'西湖区')，('四川省'，'成都市'，'武侯区')]
streets=['科技园路'，'人民大道'，'解放路'，'中山路']; levels=['普通会员'，'银卡会员'，'金卡会员'，'钻石会员']
statuses=['待付款'，'已付款'，'已发货'，'已完成'，'已取消']; channels=['APP'，'小程序'，'Web'，'门店']; pays=['支付宝'，'微信支付'，'银行卡'，'货到付款']
products=[('P1001'，'小米手机 14'，'手机数码'，'小米'，3999)，('P1004'，'戴森吹风机'，'家用电器'，'Dyson'，2999)，('P1007'，'三只松鼠坚果'，'食品生鲜'，'三只松鼠'，89)]
express=['顺丰速运'，'圆通速递'，'中通快递'，'京东物流']; warehouses=['华南仓'，'华东仓'，'华北仓'，'西南仓']

producer=KafkaProducer(**common， value_serializer=lambda v: json.dumps(v， ensure_ascii=False).encode('utf-8'))
base=datetime.datetime(2026，9，1)
for i in range(1，101):
    cust=random.choice(surnames)+random.choice(names); recv=random.choice(surnames)+random.choice(names)
    prov，city，dist=random.choice(provinces); p=random.choice(products); qty=random.randint(1，5); unit=p[4]
    subtotal=unit*qty; discount=random.choice([0，0，0，10，50，100]); freight=random.choice([0，0，8，12，15]); actual=subtotal-discount+freight
    ctime=base+datetime.timedelta(days=random.randint(0，9)，hours=random.randint(0，23)); status=random.choice(statuses)
    rec={'order_id':'ORD%08d'%i，'order_no':'SY%s%04d'%(ctime.strftime('%Y%m%d')，i)，'order_status':status，
      'order_channel':random.choice(channels)，'created_at':ctime.isoformat()，
      'paid_at':(ctime+datetime.timedelta(minutes=30)).isoformat() if status!='待付款' else None，
      'currency':'CNY'，'total_amount':subtotal，'discount_amount':discount，'freight':freight，'actual_amount':actual，
      'payment_method':random.choice(pays)，'account_id':'ACC%06d'%random.randint(1，60)，'customer_name':cust，
      'gender':random.choice(['男'，'女'])，'phone':'138%08d'%random.randint(0，99999999)，'email':'user%d@example.com'%random.randint(1000，9999)，
      'member_level':random.choice(levels)，'register_date':'2025-01-15'，'points':random.randint(0，50000)，
      'receiver_name':recv，'receiver_phone':'139%08d'%random.randint(0，99999999)，'country':'中国'，'province':prov，'city':city，'district':dist，
      'street_address':random.choice(streets)+'%d 号'%random.randint(1，999)，'postal_code':'%06d'%random.randint(100000，999999)，
      'product_id':p[0]，'product_name':p[1]，'category':p[2]，'brand':p[3]，'sku':'SKU%s%03d'%(p[0]，random.randint(1，999))，
      'quantity':qty，'unit_price':unit，'subtotal':subtotal，'warehouse':random.choice(warehouses)，
      'shipping_company':random.choice(express)，'tracking_no':'SF%013d'%random.randint(0，9999999999999)}
    producer.send(TOPIC， rec)
producer.flush(); producer.close(); print('PRODUCE_DONE 100 orders')

aws ssm send-command --region us-east-1 --instance-ids i-xxxxxxxx \
  --document-name AWS-RunShellScript \
  --parameters 'commands=["cd /opt && python3 order_producer.py"]'
# 预期: PRODUCE_DONE 100 orders

-- 外部 schema(连 MSK，IAM 认证)
CREATE EXTERNAL SCHEMA msk_schema
FROM KAFKA
IAM_ROLE 'arn:aws:iam::<ACCOUNT_ID>:role/RedshiftMSKStreamingRole'
AUTHENTICATION IAM
URI 'boot-xxxxxxxx.c2.kafka-serverless.us-east-1.amazonaws.com:9098';

-- 物化视图(自动刷新， JSON→SUPER)
CREATE MATERIALIZED VIEW mv_orders_raw AUTO REFRESH YES AS
SELECT kafka_partition， kafka_offset， kafka_timestamp，
       JSON_PARSE(kafka_value) AS payload
FROM msk_schema."orders-topic";

-- 首刷 + 验证
REFRESH MATERIALIZED VIEW mv_orders_raw;
SELECT COUNT(*) FROM mv_orders_raw;   -- 应为 100

SELECT payload.order_id::varchar   AS order_id，
       payload.member_level::varchar AS member_level，
       payload.actual_amount::int   AS actual_amount，
       payload.province::varchar    AS province
FROM mv_orders_raw ORDER BY kafka_offset;

-- 按会员等级
SELECT payload.member_level::varchar AS member_level，
       COUNT(*) AS order_cnt， SUM(payload.actual_amount::int) AS total_amount
FROM mv_orders_raw GROUP BY 1 ORDER BY total_amount DESC;

# EC2 上用 Docker 跑 Kafka UI(SSM 下发，IAM 认证)
docker run -d --name kafka-ui --restart unless-stopped -p 8080:8080 \
  -e KAFKA_CLUSTERS_0_NAME=my-msk \
  -e KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=boot-xxxxxxxx.c2.kafka-serverless.us-east-1.amazonaws.com:9098 \
  -e KAFKA_CLUSTERS_0_PROPERTIES_SECURITY_PROTOCOL=SASL_SSL \
  -e KAFKA_CLUSTERS_0_PROPERTIES_SASL_MECHANISM=AWS_MSK_IAM \
  -e "KAFKA_CLUSTERS_0_PROPERTIES_SASL_JAAS_CONFIG=software.amazon.msk.auth.iam.IAMLoginModule required;" \
  -e KAFKA_CLUSTERS_0_PROPERTIES_SASL_CLIENT_CALLBACK_HANDLER_CLASS=software.amazon.msk.auth.iam.IAMClientCallbackHandler \
  provectuslabs/kafka-ui:latest

# 本地 SSM 端口转发后访问 http://localhost:8080
aws ssm start-session --region us-east-1 --target i-xxxxxxxx \
  --document-name AWS-StartPortForwardingSession \
  --parameters '{"portNumber":["8080"]，"localPortNumber":["8080"]}'

aws iam create-role --role-name kafka-broker-ssm-role \
  --assume-role-policy-document '{"Version":"2012-10-17"，"Statement":[{"Effect":"Allow"，"Principal":{"Service":"ec2.amazonaws.com"}，"Action":"sts:AssumeRole"}]}'
aws iam attach-role-policy --role-name kafka-broker-ssm-role \
  --policy-arn arn:aws:iam::aws:policy/AmazonSSMManagedInstanceCore
aws iam create-instance-profile --instance-profile-name kafka-broker-ssm-profile
aws iam add-role-to-instance-profile --instance-profile-name kafka-broker-ssm-profile \
  --role-name kafka-broker-ssm-role

KAFKA_SG=$(aws ec2 create-security-group --region us-east-1 \
  --group-name kafka-broker-sg --description "Kafka broker SG" \
  --vpc-id vpc-xxxxxxxx --query GroupId --output text)

# 放行 9092(本地测试) 与 9094(TLS 给 Redshift)，仅来自 Redshift 安全组
aws ec2 authorize-security-group-ingress --region us-east-1 --group-id $KAFKA_SG \
  --ip-permissions \
  "IpProtocol=tcp，FromPort=9092，ToPort=9092，UserIdGroupPairs=[{GroupId=sg-redshift-xxxx}]" \
  "IpProtocol=tcp，FromPort=9094，ToPort=9094，UserIdGroupPairs=[{GroupId=sg-redshift-xxxx}]"

#!/bin/bash
set -xe
exec > /var/log/kafka-bootstrap.log 2>&1
KAFKA_VER=4.1.2; SCALA_VER=2.13; KAFKA_HOME=/opt/kafka
dnf -y install java-17-amazon-corretto-headless tar unzip
cd /opt
# 用快 CDN dlcdn(勿用慢的 archive.apache.org)
curl -fSL --retry 3 -o kafka.tgz "https://dlcdn.apache.org/kafka/${KAFKA_VER}/kafka_${SCALA_VER}-${KAFKA_VER}.tgz"
tar -xzf kafka.tgz && ln -sfn "/opt/kafka_${SCALA_VER}-${KAFKA_VER}" "$KAFKA_HOME"
mkdir -p /var/lib/kafka-data
TOKEN=$(curl -s -X PUT "http://169.254.169.254/latest/api/token" -H "X-aws-ec2-metadata-token-ttl-seconds: 21600")
PRIVATE_IP=$(curl -s -H "X-aws-ec2-metadata-token: $TOKEN" http://169.254.169.254/latest/meta-data/local-ipv4)
CONFIG="$KAFKA_HOME/config/server.properties"     # Kafka 4.x 路径
sed -i "s#^log.dirs=.*#log.dirs=/var/lib/kafka-data#" "$CONFIG"
sed -i "s#^advertised.listeners=.*#advertised.listeners=PLAINTEXT://${PRIVATE_IP}:9092#" "$CONFIG"
KID=$("$KAFKA_HOME/bin/kafka-storage.sh" random-uuid)
"$KAFKA_HOME/bin/kafka-storage.sh" format -t "$KID" -c "$CONFIG" --standalone   # 4.x 用 --standalone
cat >/etc/systemd/system/kafka.service <<EOF
[Unit]
Description=Apache Kafka (KRaft)
After=network-online.target
Wants=network-online.target
[Service]
Type=simple
ExecStart=$KAFKA_HOME/bin/kafka-server-start.sh $CONFIG
ExecStop=$KAFKA_HOME/bin/kafka-server-stop.sh
Restart=on-failure
LimitNOFILE=100000
[Install]
WantedBy=multi-user.target
EOF
systemctl daemon-reload && systemctl enable --now kafka

INSTANCE_ID=$(aws ec2 run-instances --region us-east-1 \
  --image-id ami-xxxxxxxx --instance-type t3.medium \
  --subnet-id subnet-xxxxxxxx --security-group-ids $KAFKA_SG \
  --iam-instance-profile Name=kafka-broker-ssm-profile \
  --metadata-options 'HttpEndpoint=enabled，HttpTokens=required' \
  --block-device-mappings '[{"DeviceName":"/dev/xvda"，"Ebs":{"VolumeSize":30，"VolumeType":"gp3"，"Encrypted":true，"DeleteOnTermination":true}}]' \
  --user-data file://kafka-userdata.sh \
  --tag-specifications 'ResourceType=instance，Tags=[{Key=Name，Value=kafka-broker}]' \
  --query 'Instances[0].InstanceId' --output text)

aws ssm start-session --target i-xxxxxxxx --region us-east-1
# 进入后:
sudo systemctl is-active kafka                  # 期望 active
sudo /opt/kafka/bin/kafka-topics.sh --version   # 期望 4.1.2
# 本机 9092 收发测试
B=/opt/kafka/bin; S=localhost:9092
sudo $B/kafka-topics.sh --bootstrap-server $S --create --topic kafka-stream-test --partitions 1 --replication-factor 1
echo '{"id":1，"name":"alice"，"city":"Beijing"}' | sudo $B/kafka-console-producer.sh --bootstrap-server $S --topic kafka-stream-test
sudo $B/kafka-console-consumer.sh --bootstrap-server $S --topic kafka-stream-test --from-beginning --max-messages 1

CERT_ARN=$(aws acm request-certificate --region us-east-1 \
  --domain-name kafka.example.com --key-algorithm RSA_2048 \
  --validation-method DNS --options Export=ENABLED \
  --query CertificateArn --output text)

# 取 DNS 验证记录
aws acm describe-certificate --region us-east-1 --certificate-arn $CERT_ARN \
  --query 'Certificate.DomainValidationOptions[0].ResourceRecord'

aws route53 change-resource-record-sets --hosted-zone-id ZXXXXXXXX \
  --change-batch '{"Changes":[
    {"Action":"UPSERT"，"ResourceRecordSet":{"Name":"<验证 CNAME 名>"，"Type":"CNAME"，"TTL":300，"ResourceRecords":[{"Value":"<验证 CNAME 值>"}]}}，
    {"Action":"UPSERT"，"ResourceRecordSet":{"Name":"kafka.example.com."，"Type":"A"，"TTL":60，"ResourceRecords":[{"Value":"<broker 私有 IP>"}]}}
  ]}'

# 等待 ISSUED
aws acm describe-certificate --region us-east-1 --certificate-arn $CERT_ARN --query 'Certificate.Status'

aws iam put-role-policy --role-name kafka-broker-ssm-role \
  --policy-name acm-export-kafka-cert \
  --policy-document '{"Version":"2012-10-17"，"Statement":[{"Effect":"Allow"，"Action":"acm:ExportCertificate"，"Resource":"'$CERT_ARN'"}]}'

ARN=<CERT_ARN>; REGION=us-east-1; DOMAIN=kafka.example.com
SSLDIR=/opt/kafka/ssl; CONFIG=/opt/kafka/config/server.properties

# 安装 aws cli v2(AL2023 默认无)
dnf install -y unzip >/dev/null 2>&1 || true
if ! command -v aws >/dev/null 2>&1; then
  curl -s https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip -o /tmp/awscliv2.zip
  cd /tmp && unzip -q -o awscliv2.zip && ./aws/install --update >/dev/null 2>&1
fi
export PATH=/usr/local/bin:$PATH

# 从 ACM 导出证书 → 生成 PKCS12 keystore
mkdir -p $SSLDIR && cd $SSLDIR
PP=$(openssl rand -base64 24); printf '%s' "$PP" > pp.txt
KSPASS=$(openssl rand -base64 24)
aws acm export-certificate --certificate-arn "$ARN" --passphrase fileb://pp.txt --region "$REGION" > export.json
python3 -c "import json;d=json.load(open('export.json'));open('cert.pem'，'w').write(d['Certificate']);open('chain.pem'，'w').write(d.get('CertificateChain'，'') or '');open('key.enc.pem'，'w').write(d['PrivateKey'])"
openssl pkey -in key.enc.pem -passin file:pp.txt -out key.pem
cat cert.pem chain.pem > fullchain.pem
openssl pkcs12 -export -in fullchain.pem -inkey key.pem -out kafka.keystore.p12 -passout pass:"$KSPASS" -name kafka
chmod 600 kafka.keystore.p12
rm -f export.json key.enc.pem pp.txt cert.pem chain.pem fullchain.pem key.pem

# 配置监听:保留 9092 PLAINTEXT，新增 9094 SSL
TOKEN=$(curl -s -X PUT http://169.254.169.254/latest/api/token -H 'X-aws-ec2-metadata-token-ttl-seconds: 21600')
PRIVATE_IP=$(curl -s -H "X-aws-ec2-metadata-token: $TOKEN" http://169.254.169.254/latest/meta-data/local-ipv4)
sed -i "s#^listeners=.*#listeners=PLAINTEXT://:9092，CONTROLLER://:9093，SSL://:9094#" $CONFIG
sed -i "s#^advertised.listeners=.*#advertised.listeners=PLAINTEXT://${PRIVATE_IP}:9092，SSL://${DOMAIN}:9094#" $CONFIG
cat >> $CONFIG <<EOF
ssl.keystore.location=/opt/kafka/ssl/kafka.keystore.p12
ssl.keystore.password=${KSPASS}
ssl.key.password=${KSPASS}
ssl.keystore.type=PKCS12
ssl.client.auth=none
EOF
chmod 600 $CONFIG
systemctl restart kafka

echo | openssl s_client -connect kafka.example.com:9094 -servername kafka.example.com 2>/dev/null \
  | grep -E 'subject=|issuer=|Verify return code'

-- 外部 schema(none + TLS，无需 IAM_ROLE)
CREATE EXTERNAL SCHEMA selfbuilt_kafka_ext
FROM KAFKA
URI 'kafka.example.com:9094'
AUTHENTICATION none;

-- 物化视图(topic 名含连字符，必须双引号)
CREATE MATERIALIZED VIEW selfbuilt_kafka_mv AUTO REFRESH YES AS
SELECT kafka_partition， kafka_offset， kafka_timestamp， kafka_key，
       kafka_value， JSON_PARSE(kafka_value) AS data
FROM selfbuilt_kafka_ext."kafka-stream-test"
WHERE CAN_JSON_PARSE(kafka_value);

-- 刷新 + 查询
REFRESH MATERIALIZED VIEW selfbuilt_kafka_mv;
SELECT kafka_offset， data.id::int AS id， data.name::text AS name， data.city::text AS city
FROM selfbuilt_kafka_mv ORDER BY kafka_offset;

-- 监控
SELECT * FROM SYS_STREAM_SCAN_STATES ORDER BY record_time DESC;
SELECT * FROM SYS_STREAM_SCAN_ERRORS ORDER BY record_time DESC;

SECRET=<admin-secret-arn>
aws redshift-data execute-statement --region us-east-1 \
  --workgroup-name <workgroup> --database dev --secret-arn $SECRET \
  --sql "REFRESH MATERIALIZED VIEW selfbuilt_kafka_mv;"

-- 方式一:指定角色 ARN(标准写法)
CREATE EXTERNAL SCHEMA s3t
FROM DATA CATALOG DATABASE '<namespace>'
IAM_ROLE 'arn:aws:iam::<ACCOUNT_ID>:role/RedshiftS3TablesRole'
REGION 'us-east-1'
CATALOG_ID '<ACCOUNT_ID>:s3tablescatalog/<表桶名>';

-- 方式二:IAM 联合身份直通(推荐，交由 Lake Formation 细粒度鉴权)
CREATE EXTERNAL SCHEMA s3t
FROM DATA CATALOG DATABASE '<namespace>'
CATALOG_ID '<ACCOUNT_ID>:s3tablescatalog/<表桶名>'
IAM_ROLE 'SESSION'
CATALOG_ROLE 'SESSION';

SELECT * FROM s3t.<表名>;