Conceitos básicos da ingestão de streaming do Amazon Managed Streaming para Apache Kafka

Se você usar AUTHENTICATION NONE para se conectar ao MSK, nenhum perfil do IAM será necessário. No entanto, se você usar o AUTHENTICATION IAM, deverá ter um perfil do IAM com as permissões apropriadas aplicadas. Crie um perfil do IAM com uma política de confiança que permita que o cluster do Amazon Redshift ou o grupo de trabalho do Amazon Redshift sem servidor assuma o perfil. Após a criação, o perfil deve ter permissões de política do IAM aplicadas conforme o exemplo a seguir. Essas permissões são compatíveis com a autenticação do IAM com o cluster do Amazon MSK. Para obter informações sobre como configurar a política de confiança da função do IAM, consulte Autorização do Amazon Redshift para acessar outros serviços da AWS em seu nome.

{
    "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/MyTestCluster/*",
                "arn:aws:kafka:*:0123456789:topic/MyTestCluster/*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "kafka-cluster:AlterGroup",
                "kafka-cluster:DescribeGroup"
            ],
            "Resource": [
                "arn:aws:kafka:*:0123456789:group/MyTestCluster/*"
            ]
        }
    ]
}

Verifique sua VPC e verifique se seu cluster do Amazon Redshift ou Amazon Redshift sem servidor tem uma rota para chegar ao cluster do Amazon MSK. As regras do grupo de segurança de entrada para o cluster do Amazon MSK devem dar permissão ao grupo de segurança do cluster do Amazon Redshift ou ao grupo de trabalho do Amazon Redshift sem servidor. As portas que você especifica dependem do método de autenticação usado em seu cluster, ao usar o Amazon MSK. Para obter mais informações, consulte Informações sobre portas e Acesso de dentro da AWS, mas de fora da VPC.

A autenticação do cliente com mTLS não é compatível com ingestão de streaming. Para obter mais informações, consulte Limitações.

A seguinte tabela mostra as opções de configuração complementares a serem definidas para a ingestão de streaming do Amazon MSK:

A autenticação do Amazon Redshift é definida na instrução CREATE EXTERNAL SCHEMA.

Quando o cluster do Amazon MSK tiver a autenticação Mutual Transport Layer Security (mTLS) habilitada, a configuração do Amazon Redshift para usar AUTHENTICATION NONE o direcionará a usar a porta 9094 para acesso não autenticado. No entanto, isso apresentará falha, já que a porta está sendo usada pela autenticação mTLS. Por isso, recomendamos que você mude para AUTHENTICATION IAM ao usar mTLS.

Habilite o roteamento de VPC avançado no cluster do Amazon Redshift ou no grupo de trabalho do Amazon Redshift sem servidor. Para obter mais informações, consulte Habilitar o roteamento aprimorado de VPC.

No Amazon Redshift, crie um esquema externo para mapear ao cluster do Amazon MSK. A sintaxe é a seguinte:

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

Para especificar o streaming do Amazon MSK, use FROM MSK. Para se conectar, forneça o URI do agente de bootstrap e o CLUSTER_ARN. A ingestão de streaming para o Amazon MSK fornece os seguintes tipos de autenticação quando você cria o esquema externo:

O exemplo a seguir mostra como definir o URI do agente e o CLUSTER_ARN para o cluster Amazon MSK ao criar o esquema externo:

CREATE EXTERNAL SCHEMA my_schema
FROM MSK
IAM_ROLE 'arn:aws:iam::012345678901:role/my_role'
AUTHENTICATION none
URI 'b-1.myTestCluster.123z8u.c2.kafka.us-west-1.amazonaws.com:9092,b-2.myTestCluster.123z8u.c2.kafka.us-west-1.amazonaws.com:9092'
CLUSTER_ARN 'arn:aws:kafka:us-east-1:0123456789019:cluster/SalesCluster/abcd1234-abcd-cafe-abab-9876543210ab-4';

Esse comando cria um esquema externo com propriedades de conexão para o Amazon MSK. Observe o seguinte:

Consulte mais informações sobre como criar um esquema externo em CREATE EXTERNAL SCHEMA.

Crie uma visão materializada para consumir os dados do tópico. Use um comando SQL como esse exemplo se você não quiser que os registros de erro sejam ignorados.

CREATE MATERIALIZED VIEW MyView AUTO REFRESH YES AS
SELECT *
FROM MySchema."mytopic";

O exemplo a seguir define uma exibição materializada com dados de origem JSON. Observe que a visão a seguir valida que os dados são um JSON válido e utf8. Nomes de tópico do Kafka diferenciam letras maiúsculas e minúsculas, podendo conter ambas. Para ingerir tópicos com nomes em letras maiúsculas, é possível definir a configuração enable_case_sensitive_identifier como true no nível do banco de dados. Para obter mais informações, consulte Nomes e identificadores e 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 kafka_data, 
 kafka_headers,
 refresh_time
FROM MySchema."mytopic"
WHERE CAN_JSON_PARSE(kafka_value);

Para ativar a atualização automática, use AUTO REFRESH YES. O comportamento padrão é atualização manual.

As colunas de metadados incluem o seguinte:

É importante observar que, se você tiver uma lógica de negócios na definição de visão materializada, erros nessa lógica poderão causar um bloqueio na ingestão de streaming em alguns casos. Isso pode fazer com que você precise descartar e recriar a visão materializada. Para evitar isso, recomendamos manter a simplicidade da lógica de negócios e executar lógica adicional nos dados após a ingestão.

Atualize a visão, o que invoca o Amazon Redshift para ler o tópico e carregar dados na visão materializada.

REFRESH MATERIALIZED VIEW MyView;

Consulte dados na exibição materializada.

select * from MyView;

A visão materializada é atualizada diretamente do tópico quando REFRESH é executada. Você cria uma visão materializada que mapeia à fonte de dados do tópico do Kafka. Você pode executar filtragem e agregações nos dados como parte da definição de visão materializada. Sua visão materializada de ingestão de streaming (a visão materializada de base) pode fazer referência a apenas um tópico do Kafka, mas você pode criar visões materializadas adicionais que se unam à visão materializada de base e com outras visões materializadas ou tabelas.