Utilisation du filtrage des événements avec une source d'MSKévénements Amazon

Vous pouvez utiliser le filtrage d’événements pour contrôler les enregistrements d’un flux ou d’une file d’attente que Lambda envoie à votre fonction. Pour des informations générales sur le fonctionnement du filtrage des événements, consultezContrôlez les événements que Lambda envoie à votre fonction.

Cette section se concentre sur le filtrage des événements pour les sources MSK d'événements Amazon.

Principes de base du filtrage des MSK événements Amazon

Supposons qu'un producteur rédige des messages pour un sujet de votre MSK cluster Amazon, soit dans un JSON format valide, soit sous forme de chaînes simples. Un exemple d’enregistrement ressemblerait à ce qui suit, avec le message converti en chaîne encodée en Base64 dans le champ value.

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

Supposons que votre producteur Apache Kafka rédige des messages pour votre sujet dans le JSON format suivant.

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

Vous pouvez utiliser la clé value pour filtrer les enregistrements. Supposons que vous vouliez filtrer uniquement les enregistrements où device_ID commence par les lettres AB. L’objet FilterCriteria serait le suivant.

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

Pour plus de clarté, voici la valeur du filtre Pattern expansé en clairJSON.

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

Vous pouvez ajouter votre filtre à l'aide de la console AWS CLI ou d'un AWS SAM modèle.

Console

Pour ajouter ce filtre à l’aide de la console, suivez les instructions de Attacher des critères de filtre à un mappage de sources d’événements (console) et saisissez la chaîne suivante comme critère de filtre.

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

AWS CLI

Pour créer un nouveau mappage de source d'événements avec ces critères de filtre à l'aide de AWS Command Line Interface (AWS CLI), exécutez la commande suivante.

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\" } ] } }"}]}'

Pour ajouter ces critères de filtre à un mappage des sources d’événements existant, exécutez la commande suivante.

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

AWS SAM

Pour ajouter ce filtre AWS SAM, ajoutez l'extrait suivant au YAML modèle de votre source d'événement.

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

Avec AmazonMSK, vous pouvez également filtrer les enregistrements dont le message est une chaîne simple. Supposons que vous vouliez ignorer les messages dont la chaîne est « error ». L’objet FilterCriteria se présente comme suit.

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

Pour plus de clarté, voici la valeur du filtre Pattern expansé en clairJSON.

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

Vous pouvez ajouter votre filtre à l'aide de la console AWS CLI ou d'un AWS SAM modèle.

Console

Pour ajouter ce filtre à l’aide de la console, suivez les instructions de Attacher des critères de filtre à un mappage de sources d’événements (console) et saisissez la chaîne suivante comme critère de filtre.

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

AWS CLI

Pour créer un nouveau mappage de source d'événements avec ces critères de filtre à l'aide de AWS Command Line Interface (AWS CLI), exécutez la commande suivante.

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\" ] } ] }"}]}'

Pour ajouter ces critères de filtre à un mappage des sources d’événements existant, exécutez la commande suivante.

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

AWS SAM

Pour ajouter ce filtre AWS SAM, ajoutez l'extrait suivant au YAML modèle de votre source d'événement.

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

MSKLes messages Amazon doivent être des chaînes codées en UTF -8, soit des chaînes simples, soit au JSON format. C'est parce que Lambda décode les tableaux d'octets MSK Amazon UTF en -8 avant d'appliquer des critères de filtre. Si vos messages utilisent un autre encodage, tel que UTF -16 ouASCII, ou si le format du message ne correspond pas au FilterCriteria format, Lambda traite uniquement les filtres de métadonnées. Le tableau suivant résume le comportement spécifique :

Format du du message entrant	Modèle de filtre de format pour les propriétés des messages	Action obtenue.
Chaîne de texte brut	Chaîne de texte brut	Lambda filtre en fonction de vos critères de filtre.
Chaîne de texte brut	Pas de modèle de filtre pour les propriétés des données	Lambda filtre (uniquement selon les autres propriétés de métadonnées) en fonction de vos critères de filtre.
Chaîne de texte brut	Valide JSON	Lambda filtre (uniquement selon les autres propriétés de métadonnées) en fonction de vos critères de filtre.
Valide JSON	Chaîne de texte brut	Lambda filtre (uniquement selon les autres propriétés de métadonnées) en fonction de vos critères de filtre.
Valide JSON	Pas de modèle de filtre pour les propriétés des données	Lambda filtre (uniquement selon les autres propriétés de métadonnées) en fonction de vos critères de filtre.
Valide JSON	Valide JSON	Lambda filtre en fonction de vos critères de filtre.
Chaîne non codée en UTF -8	JSON, chaîne simple ou sans motif	Lambda filtre (uniquement selon les autres propriétés de métadonnées) en fonction de vos critères de filtre.