Amazon Kinesis - Amazon Timestream
Usando Servizio gestito da Amazon per Apache FlinkUsare EventBridge Pipes

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à.

Amazon Kinesis

È possibile inviare dati da Kinesis Data Streams a LiveAnalytics Timestream per utilizzare il connettore dati di esempio per Managed Service Timestream for Apache Flink. Per ulteriori informazioni, consulta Apache Servizio gestito da Amazon per Apache Flink Flink.

Utilizzo di EventBridge Pipes per inviare dati Kinesis a Timestream

Puoi utilizzare EventBridge Pipes per inviare dati da un flusso Kinesis a un Amazon Timestream for table. LiveAnalytics

Le pipe sono destinate point-to-point alle integrazioni tra sorgenti e destinazioni supportate, con supporto per trasformazioni e arricchimenti avanzati. Le pipe riducono la necessità di conoscenze specializzate e codice di integrazione durante lo sviluppo di architetture basate sugli eventi. Per configurare una pipe, si sceglie l'origine, si aggiungono filtri facoltativi, si definisce l'arricchimento facoltativo e si sceglie la destinazione per i dati dell'evento.

Una sorgente invia gli eventi a una EventBridge pipe, che filtra e indirizza gli eventi corrispondenti alla destinazione.

Questa integrazione consente di sfruttare la potenza delle funzionalità di analisi dei dati delle serie temporali Timestream di cui dispone, semplificando al contempo la pipeline di inserimento dei dati.

L'utilizzo di EventBridge Pipes with offre i seguenti vantaggi: Timestream

  • Inserimento di dati in tempo reale: trasmetti i dati da Kinesis direttamente a Timestream per LiveAnalytics consentire analisi e monitoraggio in tempo reale.

  • Integrazione perfetta: utilizza EventBridge Pipes per gestire il flusso di dati senza la necessità di complesse integrazioni personalizzate.

  • Filtraggio e trasformazione migliorati: filtra o trasforma i record Kinesis prima che vengano archiviati Timestream per soddisfare i requisiti specifici di elaborazione dei dati.

  • Scalabilità: gestisci flussi di dati ad alta velocità e garantisci un'elaborazione efficiente dei dati con funzionalità integrate di parallelismo e batch.

Configurazione

Per configurare una EventBridge Pipe per lo streaming di dati da Kinesis a Timestream, segui questi passaggi:

  1. Creare un flusso Kinesis

    Assicurati di avere un flusso di dati Kinesis attivo da cui desideri importare i dati.

  2. Crea un Timestream database e una tabella

    Configura il Timestream database e la tabella in cui verranno archiviati i dati.

  3. Configura la EventBridge Pipe:

    • Sorgente: seleziona il tuo stream Kinesis come sorgente.

    • Obiettivo: scegli Timestream come destinazione.

    • Impostazioni di batch: definisci la finestra di batch e le dimensioni del batch per ottimizzare l'elaborazione dei dati e ridurre la latenza.

Importante

Quando si configura una tubazione, si consiglia di verificare la correttezza di tutte le configurazioni inserendo alcuni record. Tieni presente che la corretta creazione di una tubazione non garantisce che la pipeline sia corretta e che i dati scorrano senza errori. Potrebbero verificarsi errori di runtime, come una tabella errata, un parametro del percorso dinamico errato o un Timestream record non valido dopo l'applicazione della mappatura, che verranno rilevati quando i dati effettivi fluiranno attraverso la pipe.

Le seguenti configurazioni determinano la velocità con cui i dati vengono inseriti:

  • BatchSize: La dimensione massima del batch per cui verrà inviato a Timestream. LiveAnalytics Intervallo: 0 - 100. Si consiglia di mantenere questo valore su 100 per ottenere la massima produttività.

  • MaximumBatchingWindowInSeconds: Il tempo massimo di attesa per il riempimento batchSize prima che il batch venga inviato a Timestream per la destinazione. LiveAnalytics A seconda della frequenza degli eventi in arrivo, questa configurazione determinerà il ritardo di inserimento. Si consiglia di mantenere questo valore < 10 secondi per continuare a inviare i dati quasi in tempo reale. Timestream

  • ParallelizationFactor: il numero di batch da elaborare contemporaneamente da ogni shard. Si consiglia di utilizzare il valore massimo di 10 per ottenere la massima produttività e un'ingestione quasi in tempo reale.

    Se lo stream viene letto da più destinatari, utilizza il fan-out avanzato per fornire un pubblico dedicato alla tua pipe e ottenere un throughput elevato. Per ulteriori informazioni, consulta la sezione Sviluppo di utenti con fan-out avanzati nella Guida per l' Kinesis Data Streams APIutente.Kinesis Data Streams

Nota

Il throughput massimo che può essere raggiunto è limitato dalle esecuzioni simultanee di pipe per account.

La seguente configurazione garantisce la prevenzione della perdita di dati:

  • DeadLetterConfig: Si consiglia di eseguire sempre la configurazione DeadLetterConfig in modo da evitare qualsiasi perdita di dati nei casi in cui gli eventi non possano essere importati in Timestream a LiveAnalytics causa di errori dell'utente.

Ottimizza le prestazioni della tua pipe con le seguenti impostazioni di configurazione, che aiutano a evitare che i record causino rallentamenti o blocchi.

  • MaximumRecordAgeInSeconds: I record più vecchi di questo non verranno elaborati e verranno spostati direttamente in. DLQ Si consiglia di impostare questo valore in modo che non sia superiore al periodo di conservazione dell'archivio di memoria configurato per la Timestream tabella di destinazione.

  • MaximumRetryAttempts: Il numero di tentativi di riprovare un record prima che il record venga inviato a DeadLetterQueue. Si consiglia di configurarlo a 10. Questo dovrebbe aiutare a risolvere eventuali problemi temporanei e, in caso di problemi persistenti, il record verrà spostato DeadLetterQueue e sbloccato nel resto dello stream.

  • OnPartialBatchItemFailure: Per le fonti che supportano l'elaborazione parziale in batch, consigliamo di abilitarla e configurarla come AUTOMATIC _ BISECT per riprovare ulteriormente i record non riusciti prima di eliminare/inviarli a. DLQ

Esempio di configurazione

Ecco un esempio di come configurare un EventBridge Pipe per lo streaming di dati da un flusso Kinesis a una Timestream tabella:

Esempio IAM aggiornamenti delle politiche per Timestream
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "timestream:WriteRecords"
            ],
            "Resource": [
                "arn:aws:timestream:us-east-1:123456789012:database/my-database/table/my-table"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "timestream:DescribeEndpoints"
            ],
            "Resource": "*"
        }
    ]
}
Esempio Configurazione del flusso Kinesis
{
  "Source": "arn:aws:kinesis:us-east-1:123456789012:stream/my-kinesis-stream",
  "SourceParameters": {
    "KinesisStreamParameters": {
        "BatchSize": 100,
        "DeadLetterConfig": {
            "Arn": "arn:aws:sqs:us-east-1:123456789012:my-sqs-queue"
        },
       "MaximumBatchingWindowInSeconds": 5,
        "MaximumRecordAgeInSeconds": 1800,
        "MaximumRetryAttempts": 10,
        "StartingPosition": "LATEST",
       "OnPartialBatchItemFailure": "AUTOMATIC_BISECT"
    }
  }
}
Esempio Timestream configurazione del target
{
    "Target": "arn:aws:timestream:us-east-1:123456789012:database/my-database/table/my-table",
    "TargetParameters": {
        "TimestreamParameters": {
            "DimensionMappings": [
                {
                    "DimensionName": "sensor_id",
                    "DimensionValue": "$.data.device_id",
                    "DimensionValueType": "VARCHAR"
                },
                {
                    "DimensionName": "sensor_type",
                    "DimensionValue": "$.data.sensor_type",
                    "DimensionValueType": "VARCHAR"
                },
                {
                    "DimensionName": "sensor_location",
                    "DimensionValue": "$.data.sensor_loc",
                    "DimensionValueType": "VARCHAR"
                }
            ],
            "MultiMeasureMappings": [
                {
                    "MultiMeasureName": "readings",
                    "MultiMeasureAttributeMappings": [
                        {
                            "MultiMeasureAttributeName": "temperature",
                            "MeasureValue": "$.data.temperature",
                            "MeasureValueType": "DOUBLE"
                        },
                        {
                            "MultiMeasureAttributeName": "humidity",
                            "MeasureValue": "$.data.humidity",
                            "MeasureValueType": "DOUBLE"
                        },
                        {
                            "MultiMeasureAttributeName": "pressure",
                            "MeasureValue": "$.data.pressure",
                            "MeasureValueType": "DOUBLE"
                        }
                    ]
                }
            ],
            "SingleMeasureMappings": [],
            "TimeFieldType": "TIMESTAMP_FORMAT",
            "TimestampFormat": "yyyy-MM-dd HH:mm:ss.SSS",
            "TimeValue": "$.data.time",
            "VersionValue": "$.approximateArrivalTimestamp"
        }
    }
}

trasformazione degli eventi

EventBridge Le pipe ti consentono di trasformare i dati prima che arrivino. TimestreamÈ possibile definire regole di trasformazione per modificare i Kinesis record in entrata, ad esempio cambiare i nomi dei campi.

Supponiamo che lo Kinesis stream contenga dati di temperatura e umidità. Puoi utilizzare una EventBridge trasformazione per rinominare questi campi prima di inserirli. Timestream

Best practice

Batching e buffering

  • Configura la finestra e le dimensioni del batch per bilanciare la latenza di scrittura e l'efficienza di elaborazione.

  • Utilizzate una finestra di batch per accumulare una quantità sufficiente di dati prima dell'elaborazione, riducendo il sovraccarico dovuto alla frequente presenza di piccoli batch.

Elaborazione parallela

Utilizzate l'ParallelizationFactorimpostazione per aumentare la concorrenza, in particolare per i flussi ad alta velocità. Ciò garantisce che più batch di ogni frammento possano essere elaborati contemporaneamente.

Trasformazione dei dati

Sfrutta le capacità di trasformazione di EventBridge Pipes per filtrare e migliorare i record prima di archiviarli. Timestream Questo può aiutare ad allineare i dati ai requisiti analitici.

Sicurezza

  • Assicuratevi che i IAM ruoli utilizzati per EventBridge Pipes dispongano delle autorizzazioni necessarie per leggere Kinesis e scrivere. Timestream

  • Utilizza misure di crittografia e controllo degli accessi per proteggere i dati in transito e a riposo.

Errori di debug

  • Disattivazione automatica delle tubazioni

    Le pipe verranno disattivate automaticamente in circa 2 ore se la destinazione non esiste o presenta problemi di autorizzazione

  • Throttles

    I tubi hanno la capacità di spegnersi automaticamente e riprovare fino a quando l'acceleratore non si riduce.

  • Abilitazione dei registri

    Ti consigliamo di abilitare i log a ERROR livello e includere i dati di esecuzione per ottenere maggiori informazioni sugli errori. In caso di errore, questi registri request/response sent/received conterranno from. Timestream Questo aiuta a comprendere l'errore associato e, se necessario, a rielaborare i record dopo averlo corretto.

Monitoraggio

Ti consigliamo di impostare allarmi su quanto segue per rilevare eventuali problemi con il flusso di dati:

  • Età massima del record nella fonte

    • GetRecords.IteratorAgeMilliseconds

  • Metriche dei guasti in Pipes

    • ExecutionFailed

    • TargetStageFailed

  • Timestream Errori di scrittura API

    • UserErrors

Per ulteriori metriche di monitoraggio, consulta Monitoraggio EventBridge nella Guida per l'EventBridge utente.