Verwenden der Ereignisfilterung mit einer MSK Amazon-Ereignisquelle

Sie können die Ereignisfilterung verwenden, um zu steuern, welche Datensätze aus einem Stream oder einer Warteschlange Lambda an Ihre Funktion sendet. Allgemeine Informationen zur Funktionsweise der Ereignisfilterung finden Sie unterSteuern Sie, welche Ereignisse Lambda an Ihre Funktion sendet.

Dieser Abschnitt konzentriert sich auf die Ereignisfilterung für MSK Amazon-Ereignisquellen.

Grundlagen der MSK Amazon-Ereignisfilterung

Angenommen, ein Producer schreibt Nachrichten zu einem Thema in Ihrem MSK Amazon-Cluster, entweder in einem gültigen JSON Format oder als einfache Zeichenketten. Ein Beispieldatensatz würde wie folgt aussehen, wobei die Nachricht im value-Feld in eine Base64-kodierte Zeichenfolge umgewandelt wurde.

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

Angenommen, Ihr Apache Kafka-Producer schreibt Nachrichten zu Ihrem Thema im folgenden JSON Format.

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

Sie können den value-Schlüssel verwenden, um Datensätze zu filtern. Angenommen, Sie möchten nur die Datensätze filtern, bei denen device_ID mit den Buchstaben AB beginnen. Das FilterCriteria-Objekt würde wie folgt aussehen.

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

Der besseren Übersichtlichkeit halber ist der Wert des Filters Pattern im Klartext JSON dargestellt.

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

Sie können Ihren Filter über die Konsole AWS CLI oder eine AWS SAM Vorlage hinzufügen.

Console

Um diesen Filter mithilfe der Konsole hinzuzufügen, folgen Sie den Anweisungen unter Anhängen von Filterkriterien an eine Ereignisquellenzuordnung (Konsole) und geben Sie die folgende Zeichenfolge für die Filterkriterien ein.

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

AWS CLI

Führen Sie den folgenden Befehl aus, um mithilfe von AWS Command Line Interface (AWS CLI) eine neue Ereignisquellenzuordnung mit diesen Filterkriterien zu erstellen.

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

Führen Sie den folgenden Befehl aus, um diese Filterkriterien zu einer vorhandenen Zuordnung von Ereignisquellen hinzuzufügen.

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

AWS SAM

Um diesen Filter mithilfe hinzuzufügen AWS SAM, fügen Sie der YAML Vorlage für Ihre Ereignisquelle den folgenden Ausschnitt hinzu.

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

Mit Amazon können Sie auch Datensätze filternMSK, bei denen es sich bei der Nachricht um eine einfache Zeichenfolge handelt. Angenommen, Sie möchten die Meldungen ignorieren, in denen die Zeichenfolge „error“ ist. Das FilterCriteria-Objekt würde wie folgt aussehen.

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

Der besseren Übersichtlichkeit halber ist hier der Wert des Filters einfach Pattern erweitertJSON.

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

Sie können Ihren Filter über die Konsole AWS CLI oder eine AWS SAM Vorlage hinzufügen.

Console

Um diesen Filter mithilfe der Konsole hinzuzufügen, folgen Sie den Anweisungen unter Anhängen von Filterkriterien an eine Ereignisquellenzuordnung (Konsole) und geben Sie die folgende Zeichenfolge für die Filterkriterien ein.

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

AWS CLI

Führen Sie den folgenden Befehl aus, um mithilfe von AWS Command Line Interface (AWS CLI) eine neue Ereignisquellenzuordnung mit diesen Filterkriterien zu erstellen.

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

Führen Sie den folgenden Befehl aus, um diese Filterkriterien zu einer vorhandenen Zuordnung von Ereignisquellen hinzuzufügen.

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

AWS SAM

Um diesen Filter mithilfe hinzuzufügen AWS SAM, fügen Sie der YAML Vorlage für Ihre Ereignisquelle den folgenden Ausschnitt hinzu.

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

MSKAmazon-Nachrichten müssen aus UTF -8-codierten Zeichenfolgen bestehen, entweder aus einfachen Zeichenketten oder im JSON Format. Das liegt daran, dass Lambda MSK Amazon-Byte-Arrays in UTF -8 dekodiert, bevor Filterkriterien angewendet werden. Wenn Ihre Nachrichten eine andere Kodierung verwenden, z. B. UTF -16 oderASCII, oder wenn das Nachrichtenformat nicht dem FilterCriteria Format entspricht, verarbeitet Lambda nur Metadatenfilter. In der folgenden Tabelle ist das Verhalten zusammengefasst:

-Format der eingehenden Nachricht	Filtermusterformat für Nachrichteneigenschaften	Resultierende Aktion
Einfache Zeichenfolge	Einfache Zeichenfolge	Lambda filtert basierend auf Ihren Filterkriterien.
Einfache Zeichenfolge	Kein Filtermuster für Dateneigenschaften	Lambda filtert (nur für die anderen Metadateneigenschaften) basierend auf Ihren Filterkriterien.
Einfache Zeichenfolge	Gültig JSON	Lambda filtert (nur für die anderen Metadateneigenschaften) basierend auf Ihren Filterkriterien.
Gültig JSON	Einfache Zeichenfolge	Lambda filtert (nur für die anderen Metadateneigenschaften) basierend auf Ihren Filterkriterien.
Gültig JSON	Kein Filtermuster für Dateneigenschaften	Lambda filtert (nur für die anderen Metadateneigenschaften) basierend auf Ihren Filterkriterien.
Gültig JSON	Gültig JSON	Lambda filtert basierend auf Ihren Filterkriterien.
Nicht- UTF -8-kodierte Zeichenfolge	JSON, einfache Zeichenfolge oder kein Muster	Lambda filtert (nur für die anderen Metadateneigenschaften) basierend auf Ihren Filterkriterien.