Amazon MSK 및 자체 관리형 Apache Kafka 이벤트 소스의 이벤트 필터링

이벤트 필터링을 사용해 스트림 또는 큐에서 Lambda가 함수로 보내는 레코드를 제어할 수 있어요. 이벤트 필터링이 어떻게 동작하는지에 대한 일반적인 정보는 Lambda가 함수로 보내는 이벤트 제어를 참고하세요.

참고
Amazon MSK 및 자체 관리형 Apache Kafka 이벤트 소스 매핑은 value 키에 대한 필터링만 지원해요.

출처: AWS Lambda 개발자 안내서

본문

Topics(주제)

Kafka 이벤트 필터링 기본

Kafka 이벤트 필터링 기본

프로듀서가 유효한 JSON 형식이나 일반 문자열로 Kafka 클러스터의 토픽에 메시지를 쓰고 있다고 가정해 보세요. 예제 레코드는 다음과 같을 수 있는데, 메시지는 value 필드에서 Base64로 인코딩된 문자열로 변환돼요.

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

Apache Kafka 프로듀서가 다음 JSON 형식으로 토픽에 메시지를 쓰고 있다고 가정해 보세요.

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

value 키를 사용해 레코드를 필터링할 수 있어요. device_ID가 AB로 시작하는 레코드만 필터링하고 싶다고 가정해 보세요. FilterCriteria 객체는 다음과 같아요.

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

더 명확히 하기 위해 필터의 Pattern 값을 일반 JSON으로 펼친 값은 다음과 같아요.

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

콘솔, AWS CLI 또는 AWS SAM 템플릿으로 필터를 추가할 수 있어요.

[ 콘솔 ]

콘솔로 이 필터를 추가하려면 이벤트 소스 매핑에 필터 기준 연결 (콘솔)의 지침을 따르고 Filter criteria(필터 기준)에 다음 문자열을 입력하세요.

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

[ AWS CLI ]

AWS Command Line Interface(AWS CLI)로 이 필터 기준이 있는 새 이벤트 소스 매핑을 만들려면 다음 명령을 실행하세요.

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

기존 이벤트 소스 매핑에 이 필터 기준을 추가하려면 다음 명령을 실행하세요.

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

[ AWS SAM ]

AWS SAM으로 이 필터를 추가하려면 이벤트 소스의 YAML 템플릿에 다음 스니펫을 추가하세요.

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

Kafka에서는 메시지가 일반 문자열인 레코드도 필터링할 수 있어요. 문자열이 "error"인 메시지를 무시하고 싶다고 가정해 보세요. FilterCriteria 객체는 다음과 같아요.

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