Kinesis 이벤트 소스와 이벤트 필터링 사용하기

Kinesis 이벤트 소스와 이벤트 필터링 사용하기 (Using event filtering with a Kinesis event source)

이벤트 필터링을 사용해 스트림이나 큐의 어떤 레코드를 Lambda가 함수로 보낼지 제어할 수 있어요. 이 섹션은 Kinesis 이벤트 소스의 이벤트 필터링에 중점을 둬요.

참고

Kinesis 이벤트 소스 매핑은 data 키에 대해서만 필터링을 지원해요.

출처: AWS Lambda 개발자 안내서

본문

Kinesis 이벤트 필터링 기본

프로듀서가 JSON 형식 데이터를 Kinesis 데이터 스트림에 넣는다고 가정해 봐요. 예시 레코드는 JSON 데이터가 data 필드에서 Base64 인코딩 문자열로 변환된 다음처럼 보일 거예요.

{
    "kinesis": {
        "kinesisSchemaVersion": "1.0",
        "partitionKey": "1",
        "sequenceNumber": "49590338271490256608559692538361571095921575989136588898",
        "data": "eyJSZW...BIn0=",
        "approximateArrivalTimestamp": 1545084650.987
        },
    "eventSource": "aws:kinesis",
    "eventVersion": "1.0",
    "eventID": "shardId-000000000006:***",
    "eventName": "aws:kinesis:record",
    "invokeIdentityArn": "arn:aws:iam::123456789012:role/lambda-role",
    "awsRegion": "us-east-2",
    "eventSourceARN": "arn:aws:kinesis:us-east-2:123456789012:stream/lambda-stream"
}

프로듀서가 스트림에 넣는 데이터가 유효한 JSON인 한, 이벤트 필터링으로 data 키를 사용해 레코드를 필터링할 수 있어요. 프로듀서가 다음 JSON 형식으로 Kinesis 스트림에 레코드를 넣는다고 가정해 봐요.

{
    "record": 12345,
    "order": {
        "type": "buy",
        "stock": "ANYCO",
        "quantity": 1000
        }
}

order 타입이 "buy"인 레코드만 필터링하려면 FilterCriteria 객체는 다음과 같을 거예요.

{
    "Filters": [
        {
            "Pattern": "{ \"data\" : { \"order\" : { \"type\" : [ \"buy\" ] } } }"
        }
    ]
}

명확성을 위해 필터의 Pattern 값을 평문 JSON으로 펼친 값은 다음과 같아요.

{
    "data": {
        "order": {
            "type": [ "buy" ]
            }
      }
}

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

콘솔

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

{ "data" : { "order" : { "type" : [ "buy" ] } } }

AWS CLI

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

aws lambda create-event-source-mapping \
    --function-name my-function \
    --event-source-arn arn:aws:kinesis:us-east-2:123456789012:stream/my-stream \
    --filter-criteria '{"Filters": [{"Pattern": "{ \"data\" : { \"order\" : { \"type\" : [ \"buy\" ] } } }"}]}'

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

aws lambda update-event-source-mapping \
    --uuid "a1b2c3d4-5678-90ab-cdef-11111EXAMPLE" \
    --filter-criteria '{"Filters": [{"Pattern": "{ \"data\" : { \"order\" : { \"type\" : [ \"buy\" ] } } }"}]}'

AWS SAM

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

FilterCriteria:
  Filters:
    - Pattern: '{ "data" : { "order" : { "type" : [ "buy" ] } } }'

Kinesis 소스에서 이벤트를 제대로 필터링하려면 data 필드와 data 필드의 필터 기준이 모두 유효한 JSON 형식이어야 해요. 어느 한쪽이라도 유효한 JSON 형식이 아니면 Lambda는 메시지를 버리거나 예외를 던져요. 다음 표는 구체적인 동작을 요약해요.

Kinesis 집계 레코드 필터링

Kinesis로 여러 레코드를 단일 Kinesis Data Streams 레코드로 집계해 데이터 처리량을 늘릴 수 있어요. Lambda는 Kinesis enhanced fan-out을 사용할 때만 집계 레코드에 필터 기준을 적용할 수 있어요. 표준 Kinesis로 집계 레코드를 필터링하는 것은 지원되지 않아요. enhanced fan-out을 사용할 때는 Lambda 함수의 트리거 역할을 하도록 Kinesis 전용 처리량 소비자를 구성해요. 그러면 Lambda가 집계 레코드를 필터링해 필터 기준을 충족하는 레코드만 전달해요.

더 알아보기 (Learn more)

• Lambda가 함수로 보내는 이벤트 제어
• Kinesis Data Streams enhanced fan-out과 AWS Lambda로 실시간 스트림 처리 성능 높이기