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

Filtraggio degli eventi da Amazon MSK e fonti di eventi Apache Kafka autogestite

Puoi utilizzare il filtro degli eventi per controllare quali record di uno stream o di una coda Lambda invia alla tua Event Function. Per informazioni generali sul funzionamento del filtraggio eventi, consulta Controlla quali eventi Lambda invia alla tua Event Function.

Nozioni di base sul filtraggio degli eventi di Kafka

Supponiamo che un produttore stia scrivendo messaggi su un argomento nel tuo cluster Kafka, in formato JSON valido o come stringhe semplici. Un record di esempio sarebbe simile al seguente, con il messaggio convertito in una stringa codificata Base64 nel campo value.

{
    "mytopic-0":[
        {
            "topic":"mytopic",
            "partition":0,
            "offset":15,
            "timestamp":1545084650987,
            "timestampType":"CREATE_TIME",
            "value":"SGVsbG8sIHRoaXMgaXMgYSB0ZXN0Lg==",
            "headers":[]
        }
    ]
}

Supponiamo che il produttore Apache Kafka stia scrivendo messaggi sul tuo argomento nel seguente formato JSON.

{
    "device_ID": "AB1234",
    "session":{
        "start_time": "yyyy-mm-ddThh:mm:ss",
        "duration": 162
    }
}

Puoi utilizzare la chiave value per filtrare i record. Supponiamo di voler filtrare solo i record in cui device_ID inizia con le lettere AB. L'oggetto FilterCriteria dovrebbe avere la struttura seguente.

{
    "Filters": [
        {
            "Pattern": "{ \"value\" : { \"device_ID\" : [ { \"prefix\": \"AB\" } ] } }"
        }
    ]
}

Per una maggiore chiarezza, ecco il valore del Pattern del filtro espanso in JSON semplice.

{
    "value": {
        "device_ID": [ { "prefix": "AB" } ]
      }
}

Puoi aggiungere il filtro utilizzando la console o un modello. AWS CLI AWS SAM

Console

Per aggiungere questo filtro utilizzando la console, segui le istruzioni riportate in Collegamento dei criteri di filtro a una mappatura dell'origine evento (console) e inserisci la seguente stringa per i criteri di filtraggio.

{ "value" : { "device_ID" : [ { "prefix":  "AB" } ] } }

AWS CLI

Per creare una nuova mappatura dell'origine degli eventi con questi criteri di filtro utilizzando AWS Command Line Interface (AWS CLI), esegui il comando seguente.

aws lambda create-event-source-mapping \
    --function-name my-function \
    --event-source-arn arn:aws:kafka:us-east-2:123456789012:cluster/my-cluster/b-8ac7cc01-5898-482d-be2f-a6b596050ea8 \
    --filter-criteria '{"Filters": [{"Pattern": "{ \"value\" : { \"device_ID\" : [ { \"prefix\":  \"AB\" } ] } }"}]}'

Per aggiungere questi criteri di filtraggio a una mappatura dell'origine degli eventi esistente, esegui il comando seguente.

aws lambda update-event-source-mapping \
    --uuid "a1b2c3d4-5678-90ab-cdef-11111EXAMPLE" \
    --filter-criteria '{"Filters": [{"Pattern": "{ \"value\" : { \"device_ID\" : [ { \"prefix\":  \"AB\" } ] } }"}]}'

AWS SAM

Per aggiungere questo filtro utilizzando AWS SAM, aggiungi il seguente frammento al modello YAML per la fonte dell'evento.

FilterCriteria:
  Filters:
    - Pattern: '{ "value" : { "device_ID" : [ { "prefix":  "AB" } ] } }'

Con Kafka, puoi anche filtrare i record in cui il messaggio è una stringa semplice. Supponiamo di voler ignorare i messaggi la cui stringa è "errore". L'oggetto FilterCriteria dovrebbe avere la struttura seguente.

{
    "Filters": [
        {
            "Pattern": "{ \"value\" : [ { \"anything-but\": [ \"error\" ] } ] }"
        }
    ]
}

Per una maggiore chiarezza, ecco il valore del Pattern del filtro espanso in JSON semplice.

{
    "value": [
        {
        "anything-but": [ "error" ]
        }
    ]
}

Puoi aggiungere il filtro utilizzando la console AWS CLI o un AWS SAM modello.

Console

Per aggiungere questo filtro utilizzando la console, segui le istruzioni riportate in Collegamento dei criteri di filtro a una mappatura dell'origine evento (console) e inserisci la seguente stringa per i criteri di filtraggio.

{ "value" : [ { "anything-but": [ "error" ] } ] }

AWS CLI

Per creare una nuova mappatura dell'origine degli eventi con questi criteri di filtro utilizzando AWS Command Line Interface (AWS CLI), esegui il comando seguente.

aws lambda create-event-source-mapping \
    --function-name my-function \
    --event-source-arn arn:aws:kafka:us-east-2:123456789012:cluster/my-cluster/b-8ac7cc01-5898-482d-be2f-a6b596050ea8 \
    --filter-criteria '{"Filters": [{"Pattern": "{ \"value\" : [ { \"anything-but\": [ \"error\" ] } ] }"}]}'

Per aggiungere questi criteri di filtraggio a una mappatura dell'origine degli eventi esistente, esegui il comando seguente.

aws lambda update-event-source-mapping \
    --uuid "a1b2c3d4-5678-90ab-cdef-11111EXAMPLE" \
    --filter-criteria '{"Filters": [{"Pattern": "{ \"value\" : [ { \"anything-but\": [ \"error\" ] } ] }"}]}'

AWS SAM

Per aggiungere questo filtro utilizzando AWS SAM, aggiungi il seguente frammento al modello YAML per la fonte dell'evento.

FilterCriteria:
  Filters:
    - Pattern: '{ "value" : [ { "anything-but": [ "error" ] } ] }'

I messaggi Kafka devono essere stringhe UTF-8 codificate, semplici o in formato JSON. Questo perché Lambda decodifica gli array di byte Kafka prima di applicare i criteri di filtro. UTF-8 Se i tuoi messaggi utilizzano un'altra codifica, ad esempio ASCII, UTF-16 o se il formato del messaggio non corrisponde al formato, Lambda elabora solo i FilterCriteria filtri dei metadati. La tabella seguente riepiloga il comportamento specifico: