Ereignisfilterung mit einer selbstverwalteten Apache-Kafka-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 über die Funktionsweise der Ereignisfilterung finden Sie unter Steuern Sie, welche Ereignisse Lambda an Ihre Funktion sendet.

Dieser Abschnitt konzentriert sich auf die Ereignisfilterung für selbstverwaltete Apache Kafka-Ereignisquellen.

Grundlagen der selbstverwalteten Apache-Kafka-Ereignisfilterung

Angenommen, ein Producer schreibt Nachrichten zu einem Thema in Ihrem selbstverwalteten Apache-Kafka-Cluster, entweder im 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\" } ] } }"
        }
    ]
}

Zur Verdeutlichung sehen Sie hier den Wert des Filter-Pattern in reinem JSON.

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

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

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

Mit selbstverwaltetem Apache Kafka können Sie auch Datensätze filtern, bei denen die Nachricht eine einfache Zeichenfolge ist. 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\" ] } ] }"
        }
    ]
}

Zur Verdeutlichung sehen Sie hier den Wert des Filter-Pattern in reinem JSON.

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

Sie können Ihren Filter mithilfe der Konsole AWS CLI oder einer Vorlage hinzufügen. AWS SAM

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

Selbstverwaltete Apache-Kafka-Nachrichten müssen UTF-8-kodierte Strings sein, entweder einfache Strings oder im JSON-Format. Das liegt daran, dass Lambda Byte-Arrays von Kafka vor Anwendung der Filterkriterien in UTF-8 dekodiert. Wenn Ihre Nachrichten eine andere Kodierung nutzen, z. B. UTF-16 oder ASCII, oder das Nachrichtenformat nicht dem FilterCriteria-Format entspricht, verarbeitet Lambda nur Metadatenfilter. In der folgenden Tabelle ist das Verhalten zusammengefasst: