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

使用允许 Amazon Redshift 集群代入该角色的信任策略创建 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 MSK 集群的路由。您的 Amazon MSK 集群的入站安全组规则应允许您的 Amazon Redshift 集群的安全组。如果您使用 Amazon MSK，则您指定的端口取决于集群上使用的身份验证方法。有关更多信息，请参阅端口信息和从 Amazon 内但在 VPC 外部访问。

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

在 Amazon Redshift 集群中启用增强型 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。Kinesis 主题名称区分大小写，可以包含大写字母和小写字母。要使用区分大小写的标识符，可以在会话或集群级别将配置 enable_case_sensitive_identifier 设置为 true。有关更多信息，请参阅名称和标识符和 r_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;

要开启自动刷新，请使用 AUTO REFRESH YES。原定设置行为是手动刷新。

元数据列包括以下内容：

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

REFRESH MATERIALIZED VIEW MyView;

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

select * from MyView;

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