Lambda过滤Kinesis事件失败求助：无法排除CONTROL_MESSAGE

你的判断完全正确——Lambda事件过滤失效的核心原因就是Kinesis事件中的data字段是Base64编码的Gzip压缩内容，而Lambda的事件过滤机制只会对事件的原始JSON结构做匹配，不会自动解码、解压data字段里的内容，所以你直接针对data.messageType设置的过滤规则根本无法命中实际的消息内容。

验证依据 #

你提供的示例事件里，data字段的字符串是Base64编码的Gzip数据，解码并解压后才能看到真实的消息结构（包含messageType字段），但Lambda过滤逻辑看不到这层解析后的内容，自然无法生效。

两种可行解决方案 #

方案1：调整消息发送逻辑，让过滤字段可直接被Lambda识别 #

如果可以修改Kinesis消息的生产端，建议把messageType字段放在**Kinesis记录的属性（Record Attributes）**或者partitionKey中（如果业务允许），这些字段是未编码/压缩的，Lambda可以直接基于它们设置过滤规则：

• 生产消息时添加自定义属性messageType，值为DATA_MESSAGE或CONTROL_MESSAGE
• 对应的Lambda过滤规则示例：

  { "kinesis": { "attributes": { "messageType": [ "DATA_MESSAGE" ] } } }
  

方案2：在Lambda函数内部做过滤（放弃前置过滤） #

如果无法修改生产端逻辑，只能保留data字段的压缩编码，那就把过滤逻辑移到Lambda函数内部：

1. 先对data字段做Base64解码，再解压Gzip数据，得到原始消息JSON
2. 判断messageType是否为DATA_MESSAGE，如果是则继续处理，否则直接返回（不执行后续逻辑）

示例Python代码：

import base64
import gzip
import json

def lambda_handler(event, context):
    for record in event['Records']:
        # 解码并解压data字段
        compressed_data = base64.b64decode(record['kinesis']['data'])
        decompressed_data = gzip.decompress(compressed_data)
        message = json.loads(decompressed_data)
        
        # 过滤CONTROL_MESSAGE
        if message['messageType'] != 'DATA_MESSAGE':
            continue
        
        # 此处编写业务处理逻辑
        print(f"处理有效消息: {message}")
    return {
        'statusCode': 200,
        'body': json.dumps('过滤完成')
    }

内容的提问来源于stack exchange，提问作者Ashraf Jabnoun