Amazon SQS (S3) - IAM Role (Beta)
本页介绍如何创建 Amazon SQS (S3) - IAM Role 数据源。该数据源存储访问 Amazon SQS 队列及其对应 S3 存储桶所需的配置，用于消费从 Amazon S3 投递到 SQS 的 S3 对象创建事件。
Amazon SQS (S3) - IAM Role 仅存储 SQS (S3) 导入所需的连接和授权信息。它本身不会消费消息。实际读取 SQS 消息、解析 S3 ObjectCreated 事件并将数据写入 TiDB Cloud Lake 的过程，由 Amazon SQS (S3) 集成任务 执行。
使用场景
- 集中管理 SQS (S3) 导入所需的队列 URL、Region、IAM Role 和路径作用域
- 消费 S3
ObjectCreated事件，并将对应的对象数据写入 TiDB Cloud Lake - 使用 S3 事件通知驱动数据导入，而不是仅依赖轮询 S3 路径
- 当被多个任务引用时，在一个位置统一修改 IAM Role、队列 URL 或路径作用域
创建 Amazon SQS (S3) - IAM Role
导航到 Data > Data Sources，然后点击 Create Data Source。
选择 Amazon SQS (S3) - IAM Role 作为服务类型，然后填写连接详情：
点击 Test Connectivity 验证连接。如果测试成功，点击 OK 保存数据源。
AWS 侧配置概览
在创建数据源之前，请先在你的 AWS 账户中完成以下配置：
- 创建或准备一个 SQS 标准队列。
- 配置 SQS 队列策略，允许指定的 S3 存储桶向该队列发送消息。
- 配置 S3 存储桶通知，将
ObjectCreated事件发送到 SQS 队列。 - 创建一个 IAM Role，允许 TiDB Cloud Lake 平台角色通过
sts:AssumeRole访问该角色。 - 为该 IAM Role 附加 S3 读权限和 SQS 消费权限。
- 上传一个测试对象，并确认 S3 能够将事件投递到 SQS。
请先准备以下变量。AWS_REGION 必须是 S3 存储桶和 SQS 队列所在的 Region。EXTERNAL_ID 是来自 TiDB Cloud Lake 平台控制台的组织 ID。
export AWS_REGION="<bucket-and-sqs-region>"
export AWS_ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)
export BUCKET_NAME="<your-bucket-name>"
export BUCKET_ARN="arn:aws:s3:::$BUCKET_NAME"
export QUEUE_NAME="<your-sqs-standard-queue-name>"
export ROLE_NAME="platform-s3-sqs-consumer-role"
export PREFIX="<object-key-prefix>"
export SUFFIX="<object-key-suffix>"
export PLATFORM_SETUP_ROLE_ARN="<platform-setup-role-arn>"
export PLATFORM_LOAD_ROLE_ARN="<platform-load-role-arn>"
export EXTERNAL_ID="<platform-org-id>"
第 1 步：创建或获取 SQS 标准队列
创建一个 SQS 标准队列：
aws sqs create-queue \
--region "$AWS_REGION" \
--queue-name "$QUEUE_NAME"
获取后续步骤所需的队列 URL 和队列 ARN：
export QUEUE_URL=$(
aws sqs get-queue-url \
--region "$AWS_REGION" \
--queue-name "$QUEUE_NAME" \
--query 'QueueUrl' \
--output text
)
export QUEUE_ARN=$(
aws sqs get-queue-attributes \
--region "$AWS_REGION" \
--queue-url "$QUEUE_URL" \
--attribute-names QueueArn \
--query 'Attributes.QueueArn' \
--output text
)
我们建议为每个 SQS (S3) 数据源使用专用的 SQS 标准队列。不要将同一个队列复用于其他存储桶、其他前缀/后缀作用域或其他业务事件。
第 2 步：配置 SQS 队列策略
在进行修改前，先备份当前的 SQS 属性：
aws sqs get-queue-attributes \
--region "$AWS_REGION" \
--queue-url "$QUEUE_URL" \
--attribute-names Policy QueueArn \
> "sqs-attributes.backup.$(date +%Y%m%d-%H%M%S).json"
生成 queue-policy.json，仅允许指定的 S3 存储桶发送消息：
jq -n \
--arg policyId "$QUEUE_NAME-policy" \
--arg queueArn "$QUEUE_ARN" \
--arg bucketArn "$BUCKET_ARN" \
--arg accountId "$AWS_ACCOUNT_ID" \
'{
Version: "2012-10-17",
Id: $policyId,
Statement: [
{
Sid: "AllowS3ToSendMessage",
Effect: "Allow",
Principal: {
Service: "s3.amazonaws.com"
},
Action: "sqs:SendMessage",
Resource: $queueArn,
Condition: {
ArnLike: {
"aws:SourceArn": $bucketArn
},
StringEquals: {
"aws:SourceAccount": $accountId
}
}
}
]
}' \
> queue-policy.json
应用该策略：
jq -n \
--arg policy "$(jq -c . queue-policy.json)" \
'{Policy: $policy}' \
> set-queue-attributes.json
aws sqs set-queue-attributes \
--region "$AWS_REGION" \
--queue-url "$QUEUE_URL" \
--attributes file://set-queue-attributes.json
Step 3: 配置 S3 Bucket Notification
在进行修改前，先备份当前的 bucket notification。put-bucket-notification-configuration 会替换整个 bucket notification 配置。如果该 bucket 已经有其他通知配置，请先将它们合并，再应用新的配置。
aws s3api get-bucket-notification-configuration \
--region "$AWS_REGION" \
--bucket "$BUCKET_NAME" \
> "bucket-notification.backup.$(date +%Y%m%d-%H%M%S).json"
生成 bucket-notification.json：
jq -n \
--arg id "$QUEUE_NAME" \
--arg queueArn "$QUEUE_ARN" \
--arg prefix "$PREFIX" \
--arg suffix "$SUFFIX" \
'{
QueueConfigurations: [
(
{
Id: $id,
QueueArn: $queueArn,
Events: [
"s3:ObjectCreated:*"
]
}
+
(
[
if $prefix != "" then {Name: "prefix", Value: $prefix} else empty end,
if $suffix != "" then {Name: "suffix", Value: $suffix} else empty end
] as $rules
| if ($rules | length) > 0
then {Filter: {Key: {FilterRules: $rules}}}
else {}
end
)
)
]
}' \
> bucket-notification.json
应用该配置：
aws s3api put-bucket-notification-configuration \
--region "$AWS_REGION" \
--bucket "$BUCKET_NAME" \
--notification-configuration file://bucket-notification.json
检查配置：
aws s3api get-bucket-notification-configuration \
--region "$AWS_REGION" \
--bucket "$BUCKET_NAME"
确认 QueueArn 指向目标 SQS 队列，Events 包含 s3:ObjectCreated:*，并且 FilterRules 与 TiDB Cloud Lake 数据源中配置的 Object Key Prefix / Object Key Suffix 一致。
Step 4: 创建供 TiDB Cloud Lake Assume 的 IAM Role
生成 trust-policy.json。ExternalId 是来自 TiDB Cloud Lake（platform）控制台的组织 ID。
jq -n \
--arg platformSetupRoleArn "$PLATFORM_SETUP_ROLE_ARN" \
--arg platformLoadRoleArn "$PLATFORM_LOAD_ROLE_ARN" \
--arg externalId "$EXTERNAL_ID" \
'{
Version: "2012-10-17",
Statement: [
{
Sid: "AllowPlatformSetupAssumeRole",
Effect: "Allow",
Principal: {
AWS: $platformSetupRoleArn
},
Action: "sts:AssumeRole",
Condition: {
StringEquals: {
"sts:ExternalId": $externalId
}
}
},
{
Sid: "AllowPlatformLoadAssumeRole",
Effect: "Allow",
Principal: {
AWS: $platformLoadRoleArn
},
Action: "sts:AssumeRole",
Condition: {
StringEquals: {
"sts:ExternalId": $externalId
}
}
}
]
}' \
> trust-policy.json
创建 IAM Role：
aws iam create-role \
--role-name "$ROLE_NAME" \
--assume-role-policy-document file://trust-policy.json
如果该 role 已存在，请备份并修改 trust policy：
aws iam get-role \
--role-name "$ROLE_NAME" \
--query 'Role.AssumeRolePolicyDocument' \
--output json \
> "trust-policy.backup.$(date +%Y%m%d-%H%M%S).json"
aws iam update-assume-role-policy \
--role-name "$ROLE_NAME" \
--policy-document file://trust-policy.json
Step 5: 附加 S3/SQS 权限
生成 permissions-policy.json：
jq -n \
--arg bucketArn "$BUCKET_ARN" \
--arg objectArn "$BUCKET_ARN/*" \
--arg queueArn "$QUEUE_ARN" \
'{
Version: "2012-10-17",
Statement: [
{
Sid: "S3BucketMetadataAccess",
Effect: "Allow",
Action: [
"s3:GetBucketLocation",
"s3:ListBucket"
],
Resource: $bucketArn
},
{
Sid: "S3ObjectReadAccess",
Effect: "Allow",
Action: [
"s3:GetObject"
],
Resource: $objectArn
},
{
Sid: "SQSConsumeAccess",
Effect: "Allow",
Action: [
"sqs:ReceiveMessage",
"sqs:DeleteMessage",
"sqs:GetQueueAttributes",
"sqs:ChangeMessageVisibility"
],
Resource: $queueArn
}
]
}' \
> permissions-policy.json
应用权限：
aws iam put-role-policy \
--role-name "$ROLE_NAME" \
--policy-name platform-s3-sqs-access \
--policy-document file://permissions-policy.json
权限检查清单：
- SQS 权限的作用域限定为目标队列 ARN。
- S3 权限的作用域限定为目标 bucket 和对象 ARN。
- 默认情况下，此策略不需要 S3 写入或删除权限。
- 如果未来的 SQS（S3）集成任务启用了 PURGE 或 Clean Up Original Files，即在成功导入后删除源对象，则需要对目标对象路径授予
s3:DeleteObject权限。
第 6 步：验证 S3 到 SQS
上传一个与 PREFIX / SUFFIX 匹配的测试对象：
echo 'a,b' > /tmp/sqs-s3-local-test.csv
aws s3 cp /tmp/sqs-s3-local-test.csv \
"s3://$BUCKET_NAME/${PREFIX}sqs-s3-local-test-$(date +%s)$SUFFIX" \
--region "$AWS_REGION"
从 SQS 接收一条消息：
aws sqs receive-message \
--region "$AWS_REGION" \
--queue-url "$QUEUE_URL" \
--max-number-of-messages 1 \
--wait-time-seconds 10 \
--visibility-timeout 60
确认消息中包含 Records，eventSource 为 aws:s3，eventName 为 ObjectCreated:*，并且 Records[].s3.bucket.name 和 Records[].s3.object.key 与测试对象一致。
提供给 TiDB Cloud Lake 的信息
完成 AWS 侧配置后，在 TiDB Cloud Lake 中创建数据源时填写以下信息：
获取 role_arn 的命令示例：
aws iam get-role \
--role-name "$ROLE_NAME" \
--query 'Role.Arn' \
--output text
配置要求
- S3 存储桶和 SQS 队列应位于同一个 AWS Region。
- SQS 队列必须是标准队列。不支持 FIFO 队列。
- SQS 队列应专用于一条 S3 notification rule。不要将其复用于其他存储桶、其他 prefix / suffix 作用域或其他业务事件。
- S3 notification 中的 bucket、prefix 和 suffix 应与 TiDB Cloud Lake 数据源配置保持一致。
put-bucket-notification-configuration会替换整个存储桶 notification 配置。应用更改前，请先备份并合并现有配置。- S3 event notifications 和 SQS 标准队列都采用至少一次投递，因此消息可能会重复。
后续步骤
创建此数据源后，你可以使用它来创建 Amazon SQS (S3) 集成任务。