开始使用 Amazon Managed Streaming for Apache Kafka 串流摄取

使用允许 Amazon Redshift 集群或 Amazon Redshift Serverless 代入 IAM 角色的信任策略创建该角色。有关如何为 IAM 角色配置信任策略的信息，请参阅授权 Amazon Redshift 代表您访问其他 Amazon 服务。创建角色后，它应具有以下 IAM 策略，从而提供与 Amazon MSK 集群进行通信的权限。如果您使用 Amazon MSK，则所需的策略取决于集群上使用的身份验证方法。有关 Amazon MSK 中可用的身份验证方法，请参阅 Apache Kafka API 的身份验证和授权。

使用未经身份验证的访问权限时的 Amazon MSK IAM 策略：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor0",
            "Effect": "Allow",
            "Action": [
                "kafka:GetBootstrapBrokers"
            ],
            "Resource": "*"
        }
    ]
}

使用 IAM 身份验证时的 Amazon MSK IAM 策略：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "MSKIAMpolicy",
            "Effect": "Allow",
            "Action": [
                "kafka-cluster:ReadData",
                "kafka-cluster:DescribeTopic",
                "kafka-cluster:Connect"
            ],
            "Resource": [
                "arn:aws:kafka:*:0123456789:cluster/*/*",
                "arn:aws:kafka:*:0123456789:topic/*/*/*"
            ]
        },
        {
            "Sid": "MSKPolicy",
            "Effect": "Allow",
            "Action": [
                "kafka:GetBootstrapBrokers"
            ],
            "Resource": "*"
        }
    ]
}

检查您的 VPC，并确认您的 Amazon Redshift 集群或 Amazon Redshift Serverless 拥有通往 Amazon MSK 集群的路由。Amazon MSK 集群的入站安全组规则应允许 Amazon Redshift 集群或 Amazon Redshift Serverless 工作组的安全组。如果您使用 Amazon MSK，则您指定的端口取决于集群上使用的身份验证方法。有关更多信息，请参阅端口信息和从 Amazon 内但在 VPC 外部访问。

请注意，对于串流摄取，不支持通过 mTLS 进行客户端身份验证。有关更多信息，请参阅限制。

下表显示了为从 Amazon MSK 进行串流摄取所要设置的免费配置选项：

Amazon Redshift 身份验证是在 CREATE EXTERNAL SCHEMA 语句中设置的。

如果 Amazon MSK 集群启用了相互传输层安全性协议 (mTLS) 身份验证，则将 Amazon Redshift 配置为使用 AUTHENTICATION NONE 会指示它使用端口 9094 进行未经身份验证的访问。但是，由于 mTLS 身份验证正在使用该端口，因此这一过程将失败。因此，当您使用 mTLS 时，我们建议您切换到 AUTHENTICATION IAM。

在 Amazon Redshift 集群或 Amazon Redshift Serverless 工作组中启用增强型 VPC 路由。有关更多信息，请参阅启用增强型 VPC 路由。

在 Amazon Redshift 中，创建一个外部 Schema 以映射到 Amazon MSK 集群。

CREATE EXTERNAL SCHEMA MySchema
FROM MSK
IAM_ROLE { default | 'iam-role-arn' }
AUTHENTICATION { none | iam }
CLUSTER_ARN 'msk-cluster-arn';

在 FROM 子句中，Amazon MSK 表示模式映射来自托管式 Kafka 服务的数据。

当您创建外部模式时，Amazon MSK 的串流摄取提供以下身份验证类型：

串流摄取不支持其他 Amazon MSK 身份验证方法，例如 TLS 身份验证或用户名和密码。

CLUSTER_ARN 指定要从中进行流式传输的 Amazon MSK 集群。

创建一个实体化视图以使用来自主题的数据。以下示例定义了一个包含 JSON 源数据的实体化视图。请注意，以下视图会确认数据是有效 JSON 和 utf8。Kafka 主题名称区分大小写，可以包含大写字母和小写字母。要从具有大写名称的主题中摄取，可以在数据库级别将配置 enable_case_sensitive_identifier 设置为 true。有关更多信息，请参阅名称和标识符和 enable_case_sensitive_identifier。

CREATE MATERIALIZED VIEW MyView AUTO REFRESH YES AS
SELECT kafka_partition, 
 kafka_offset, 
 kafka_timestamp_type, 
 kafka_timestamp, 
 kafka_key, 
 JSON_PARSE(kafka_value) as Data, 
 kafka_headers
FROM MySchema."mytopic"
WHERE CAN_JSON_PARSE(kafka_value);

要开启自动刷新，请使用 AUTO REFRESH YES。默认行为是手动刷新。

元数据列包括以下内容：

请务必注意，如果您的实体化视图定义中有业务逻辑，那么在某些情况下，解析错误可能会导致串流摄取被阻止。这可能会导致您不得不删除实体化视图，然后重新创建它。为避免这种情况，我们建议您尽可能简化解析逻辑，并在摄取数据后对数据进行大部分业务逻辑检查。以下示例显示了如何使用 CAN_JSON_PARSE 函数 以防出现错误并更成功地摄取数据。

刷新视图，这会调用 Amazon Redshift 从主题中读取数据并将数据加载到实体化视图中。

REFRESH MATERIALIZED VIEW MyView;

在实体化视图中查询数据。

select * from MyView;

当 REFRESH 运行时，直接从主题更新实体化视图。您创建映射到 Kafka 主题数据来源的实体化视图。在实体化视图定义中，您可以对数据执行筛选和聚合。串流摄取实体化视图（基本实体化视图）只能引用一个 Kafka 主题，但是您可以创建额外的实体化视图，以与基本实体化视图和其他实体化视图或表连接使用。