Nozioni di base sull’importazione in streaming da origini Apache Kafka - Amazon Redshift
Configurazione dell’importazione in streaming da Kafka

Amazon Redshift non supporterà più la creazione di nuovi Python a UDFs partire dal 1° novembre 2025. Se vuoi usare Python UDFs, crea la UDFs data precedente a quella data. Python esistente UDFs continuerà a funzionare normalmente. Per ulteriori informazioni, consulta il post del blog.

Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.

Nozioni di base sull’importazione in streaming da origini Apache Kafka

In questo argomento viene descritto come utilizzare i dati in streaming di Amazon MSK, Apache Kafka o Confluent Cloud usando una vista materializzata.

Lo scopo dell'importazione dati in streaming di Amazon Redshift è semplificare il processo di importazione diretta di dati di flusso da un servizio di streaming in Amazon Redshift o in Amazon Redshift serverless. È compatibile con Amazon MSK con provisioning e Amazon MSK serverless, con Apache Kafka open source e con Confluent Cloud. L’importazione in streaming di Amazon Redshift elimina la necessità di gestire un argomento Apache Kafka in Amazon S3 prima di importare i dati di flusso in Redshift.

A livello tecnico, l’importazione in streaming fornisce l’importazione a bassa latenza e ad alta velocità dei dati del flusso o dell’argomento in una vista materializzata di Amazon Redshift. Dopo la configurazione, utilizzando l'aggiornamento della vista materializzata è possibile acquisire grandi volumi di dati.

Devi disporre di un’origine Apache Kafka prima di configurare l’importazione in streaming di Amazon Redshift. Se non disponi di un’origine, creane una utilizzando le istruzioni seguenti:

Configurazione dell’importazione in streaming da Kafka

Utilizza le seguenti procedure per configurare l'importazione di streaming su Amazon Redshift da Amazon MSK o da fonti Apache Kafka non gestite (Apache Kafka e AWS Confluent Cloud).

Configurare l’autenticazione

In questa sezione viene descritta la configurazione dell’autenticazione per consentire all’applicazione Amazon Redshift di accedere a un’origine Amazon MSK.

Dopo avere creato il ruolo dell’applicazione, collega una delle seguenti policy per consentire l’accesso al cluster Amazon MSK, Apache Kafka o Confluent Cloud. Per l’autenticazione mTLS puoi archiviare i certificati utilizzati da Amazon Redshift in ACM o Secrets Manager, quindi devi scegliere la policy che corrisponde al luogo in cui è archiviato il certificato.

Tieni presente che i certificati autofirmati non sono supportati per l’autenticazione o i dati in transito quando utilizzi l’importazione in streaming diretta in Amazon Redshift con una delle origini di streaming Apache Kafka supportate. Sono inclusi Amazon MSK, Apache Kafka e Confluent Cloud. Prendi in considerazione l'utilizzo di certificati generati con o con qualsiasi altra autorità di certificazione pubblicamente attendibile. AWS Certificate Manager

L’autenticazione IAM di Amazon Redshift con MSK è supportata solo in Kafka 2.7.1 o versione successiva.

AUTHENTICATION IAM (solo Amazon MSK):

JSON
{
    "Version":"2012-10-17",		 	 	 
    "Statement": [
        {
            "Sid": "MSKIAMpolicy",
            "Effect": "Allow",
            "Action": [
                "kafka-cluster:ReadData",
                "kafka-cluster:DescribeTopic",
                "kafka-cluster:Connect"
            ],
            "Resource": [
                "arn:aws:kafka:*:111122223333:cluster/MyTestCluster/*",
                "arn:aws:kafka:*:111122223333:topic/MyTestCluster/*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "kafka-cluster:AlterGroup",
                "kafka-cluster:DescribeGroup"
            ],
            "Resource": [
                "arn:aws:kafka:*:111122223333:group/MyTestCluster/*"
            ]
        }
    ]
}

AUTENTICAZIONE MTLS: utilizzo di un certificato memorizzato in AWS Certificate Manager

JSON
{
    "Version":"2012-10-17",		 	 	 
    "Statement": [
        {
            "Sid": "MSKmTLSACMpolicy",
            "Effect": "Allow",
            "Action": [
                "acm:ExportCertificate" 
            ],
            "Resource": [
                "arn:aws:acm:us-east-1:444455556666:certificate/certificate_ID"
            ]
        }
    ]
}

AUTENTICAZIONE MTLS: utilizzo di un certificato memorizzato in Gestione dei segreti AWS

JSON
{
    "Version":"2012-10-17",		 	 	 
    "Statement": [
        {
            "Sid": "MSKmTLSSecretsManagerpolicy",
            "Effect": "Allow",
            "Action": [
                "secretsmanager:GetSecretValue" 
            ],
            "Resource": [
                "arn:aws:secretsmanager:us-east-1:444455556666:secret:secret_ID"
            ]
        }
    ]
}
Amazon MSK

Se utilizzi AUTHENTICATION NONE per connetterti a un’origine Amazon MSK, non è richiesto alcun ruolo IAM. Tuttavia, se utilizzi AUTHENTICATION IAM o MTLS per l’autenticazione con il cluster Amazon MSK, il cluster Amazon Redshift o il namespace Amazon Redshift serverless devono avere un ruolo IAM collegato con le autorizzazioni appropriate. Crea un ruolo IAM con una policy di attendibilità che consente al cluster Amazon Redshift o al namespace Amazon Redshift serverless di assumere il ruolo. Dopo avere creato il ruolo, aggiungi una delle autorizzazioni seguenti per supportare IAM o MTLS. Per l'autenticazione MTLS, i certificati utilizzati da Amazon Redshift possono essere archiviati AWS Certificate Manager in Gestione dei segreti AWS oppure, è necessario scegliere la politica che corrisponde al luogo in cui è archiviato il certificato. Collega il ruolo al cluster con provisioning Amazon Redshift o al namespace Redshift serverless. Per informazioni su come configurare la policy di fiducia per il ruolo IAM, consulta Autorizzazione di Amazon Redshift ad accedere ad AWS altri servizi per tuo conto.

La tabella seguente mostra le opzioni di configurazione aggiuntive da impostare per l'acquisizione di streaming da Amazon MSK:

Configurazione di Amazon Redshift Configurazione di Amazon MQ Porta da aprire tra Redshift e Amazon MSK
AUTHENTICATION NONE Trasporto TLS disabilitato 9092
AUTHENTICATION NONE Trasporto TLS abilitato 9094
AUTHENTICATION IAM IAM 9098/9198
AUTHENTICATION MTLS Trasporto TLS abilitato 9094

L'autenticazione Amazon Redshift è impostata nell'istruzione CREATE EXTERNAL SCHEMA.

Nota

Nel caso in cui il cluster Amazon MSK ha l'autenticazione Mutual Transport Layer Security (mTLS) abilitata, la configurazione di Amazon Redshift per l'utilizzo di AUTHENTICATION NONE indica ad Amazon Redshift di utilizzare la porta 9094 per l'accesso non autenticato. Tuttavia, poiché la porta viene utilizzata dall'autenticazione mTLS, questa operazione avrà esito negativo. Per questo motivo consigliamo di passare ad AUTHENTICATION MTLS quando utilizzi mTLS.

Apache Kafka or Confluent Cloud

Per Apache Kafka e Confluent Cloud, Amazon Redshift supporta i seguenti protocolli di connessione:

  • Puoi utilizzare mTLS o testo semplice con il trasporto TLS per l’autenticazione quando ti connetti ad Apache Kafka.

  • Puoi utilizzare mTLS per l’autenticazione solo quando ti connetti a Confluent Cloud.

Amazon Redshift supporta i seguenti protocolli di crittografia per la connessione ad Apache Kafka o Confluent Cloud:

Metodi di autenticazione supportati per Apache Kafka e Confluent Cloud

Amazon Redshift Protocollo di sicurezza Kafka Supporto di Apache Kafka Supporto di Confluent Cloud
AUTHENTICATION NONE PLAINTEXT No No
AUTHENTICATION NONE SSL Sì No
AUTHENTICATION IAM SASL_SSL No No
AUTHENTICATION MTLS SSL Sì (con certificato) Sì (con certificato)

Tieni presente che Amazon Redshift non supporta SASL/SCRAM SASL/PLAINTEXT.

Configurazione del VPC

Dopo avere creato le risorse di autenticazione, controlla il VPC e verifica che il cluster Amazon Redshift o il gruppo di lavoro Amazon Redshift serverless disponga di un percorso per raggiungere l’origine Apache Kafka.

Nota

Per Amazon MSK, le regole in entrata del gruppo di sicurezza per il cluster Amazon MSK dovrebbero autorizzare il gruppo di sicurezza del cluster Amazon Redshift o del gruppo di lavoro Amazon Redshift serverless. Le porte specificate dipendono dai metodi di autenticazione utilizzati per il cluster Amazon MSK. Per ulteriori informazioni, consulta Informazioni sulle porte e Accesso dall'interno AWS ma dall'esterno del VPC.

Quindi abilita il routing VPC avanzato nel cluster Amazon Redshift o nel gruppo di lavoro Amazon Redshift serverless. Per ulteriori informazioni, consulta Abilitazione del routing VPC avanzato.

Creare una vista materializzata

In questa sezione configuri la vista materializzata utilizzata da Amazon Redshift per accedere ai dati in streaming di Apache Kafka.

Supponendo che abbia a disposizione un cluster Apache Kafka, il primo passo consiste nel definire uno schema in Redshift con CREATE EXTERNAL SCHEMA e nel fare riferimento al cluster come origine dati. Quindi, definisci lo STREAM in una vista materializzata per accedere ai dati nell'argomento. Puoi archiviare i record dell’argomento utilizzando il tipo di dati VARBYTE di Amazon Redshift o definire uno scherma che converte i dati nel formato SUPER semistrutturato. Quando si esegue una query sulla vista materializzata, i record restituiti sono una point-in-time visualizzazione dell'argomento.

  1. In Amazon Redshift crea uno schema esterno da mappare al cluster Apacke Kafka. La sintassi è la seguente:

    CREATE EXTERNAL SCHEMA MySchema
FROM KAFKA
[ IAM_ROLE [ default | 'iam-role-arn' ] ]
AUTHENTICATION [ none | iam | mtls ]
{AUTHENTICATION_ARN 'acm-certificate-arn' |  SECRET_ARN 'asm-secret-arn'};

    Nella clausola FROM, KAFKA indica che lo schema mappa i dati da un’origine Apache Kafka.

    AUTHENTICATION indica il tipo di autenticazione definito per l’importazione in streaming. Sono disponibili tre tipi:

    • none: specifica che non è richiesta alcuna autenticazione. Ciò corrisponde all’accesso non autenticato in MSK. Ciò corrisponde all’autenticazione SSL in Apache Kafka. Questo metodo di autenticazione non è supportato per Confluent Cloud.

    • iam: specifica l'autenticazione IAM. Puoi utilizzare l’autenticazione IAM solo con Amazon MSK. Quando scegli questa opzione, assicurati che il ruolo IAM disponga delle autorizzazioni per l'autenticazione IAM. Per ulteriori informazioni sulla configurazione delle policy IAM richieste, consulta Configurazione dell’importazione in streaming da Kafka.

    • mtls: specifica che il protocollo Transport Layer Security (TLS) reciproco fornisce comunicazioni sicure facilitando l’autenticazione tra client e server. In questo caso il client è Redshift e il server è Apache Kafka. Per ulteriori informazioni sulla configurazione dell’importazione in streaming con mTLS, consulta Autenticazione con mTLS per l’importazione in streaming Redshift da origini Apache Kafka.

    Tieni presente che l’autenticazione Amazon MSK con un nome utente e una password non è supportata per l’importazione in streaming.

    Il parametro AUTHENTICATION_ARN specifica l’ARN del certificato del protocollo Transport Layer Security reciproco (mTLS) ACM utilizzato per stabilire una connessione crittografata.

    Il SECRET_ARN parametro specifica l'arn del Gestione dei segreti AWS segreto contenente il certificato che deve essere utilizzato da Amazon Redshift per MTL.

    Negli esempi seguenti viene illustrato come impostare l’URI del broker per il cluster Amazon MSK quando crei lo schema esterno:

    Utilizzo dell’autenticazione IAM:

    CREATE EXTERNAL SCHEMA my_schema
FROM KAFKA
IAM_ROLE 'arn:aws:iam::012345678901:role/my_role'
AUTHENTICATION IAM
URI 'b-1.myTestCluster.123z8u.c2.kafka.us-west-1.amazonaws.com:9098,b-2.myTestCluster.123z8u.c2.kafka.us-west-1.amazonaws.com:9098'

    Utilizzo di nessuna autenticazione:

    CREATE EXTERNAL SCHEMA my_schema
FROM KAFKA 
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'

    Utilizzo di mTLS:

    CREATE EXTERNAL SCHEMA my_schema
FROM KAFKA
IAM_ROLE 'arn:aws:iam::012345678901:role/my_role'
AUTHENTICATION MTLS
URI 'b-1.myTestCluster.123z8u.c2.kafka.us-west-1.amazonaws.com:9094,b- 2.myTestCluster.123z8u.c2.kafka.us-west-1.amazonaws.com:9094'
{AUTHENTICATION_ARN 'acm-certificate-arn' |  SECRET_ARN 'asm-secret-arn'}

    Per informazioni su come creare uno schema esterno, consulta CREATE EXTERNAL SCHEMA.

  2. Crea una vista materializzata per consumare i dati dall'argomento. Utilizza un comando SQL come l’esempio seguente.

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

    I nomi degli argomenti Kafka distinguono tra maiuscole e minuscole e possono contenere lettere maiuscole e minuscole. Per importare da argomenti con nomi in maiuscolo, puoi impostare la configurazione enable_case_sensitive_identifier su true a livello di sessione o database. Per ulteriori informazioni, consulta Nomi e identificatori e enable_case_sensitive_identifier.

    Per attivare l'aggiornamento automatico, usa AUTO REFRESH YES. Il comportamento predefinito prevede l'aggiornamento manuale.

  3. Le colonne di metadati includono quanto segue:

    Colonna di metadati Tipo di dati Description
    kafka_partition bigint ID della partizione del record dall'argomento Kafka
    kafka_offset bigint Offset del record nell'argomento di Kafka per una determinata partizione
    kafka_timestamp_type char(1)

    Tipo di timestamp utilizzato nel record Kafka:

    • C: registra l'ora di creazione (CREATE_TIME) sul lato client

    • L: registra l'ora di aggiunta (LOG_APPEND_TIME) sul lato server Kafka

    • U: l'ora di creazione del record non è disponibile (NO_TIMESTAMP_TYPE)
    kafka_timestamp timestamp without time zone Il formato del valore timestamp per il record
    kafka_key varbyte La chiave del record Kafka
    kafka_value varbyte Il record ricevuto da Kafka
    kafka_headers super L'intestazione del record ricevuto da Kafka
    ora di aggiornamento timestamp without time zone L'ora di inizio dell'aggiornamento.

    È importante notare che, se la definizione di vista materializzata include una logica aziendale che causa errori di logica aziendale, in alcuni casi ciò può comportare errori nell’importazione in streaming. Ciò potrebbe comportare la necessità di rimuovere e ricreare la vista materializzata. Per evitare ciò, consigliamo di mantenere la logica aziendale semplice ed eseguire una logica aggiuntiva sui dati dopo l’importazione.

  4. Aggiorna la vista per invocare Amazon Redshift ed eseguire la lettura dall'argomento e il caricamento dei dati nella vista materializzata.

    REFRESH MATERIALIZED VIEW MyView;

  5. Esegui una query sui dati nella vista materializzata.

    select * from MyView;

    La vista materializzata viene aggiornata direttamente dall'argomento quando si esegue il comando REFRESH. Viene creata una vista materializzata che corrisponde all'origine dati dell'argomento Kafka. È possibile eseguire filtri e aggregazioni sui dati nell'ambito della definizione della vista materializzata. La vista materializzata dell'importazione dati in streaming (vista materializzata di base) può fare riferimento a un solo argomento Kafka, ma è possibile creare viste materializzate aggiuntive che si uniscono alla vista materializzata di base e con altre viste materializzate o tabelle.

Per ulteriori informazioni sulle limitazioni relative all'importazione dati in streaming, consulta Comportamento dell’importazione in streaming e tipi di dati.