我可以通过Lambda函数使用AWS MSK kafka主题吗？

问题描述

我认为使用lambda消耗进入AWS MSK Kafka集群中某个主题的消息是微不足道的，但是我找不到从AWS文档中做到这一点的方法。有可能吗？

解决方法

是的。您可以在Lambda代码中使用Kafka客户端库。在下面找到Python的示例：

#!/bin/env python

import json
import logging
import time
import os
from kafka import KafkaConsumer               


"""
Read data from MSK and console it out.

Required environment variables:
MSK_BOOTSTRAP_SRV: MSK Bootstrap servers. 
MSK_TOPIC_NAME: MSK topic. 

"""
kafka_client = None
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)

def lambda_handler(event,context):
    logger.debug('MSK consumer starts.')
    try:
        consumer = KafkaConsumer(os.environ['MSK_TOPIC_NAME'],auto_offset_reset='earliest',bootstrap_servers=os.environ['MSK_BOOTSTRAP_SRV'],api_version=(0,10),consumer_timeout_ms=1000)
        
        for msg in consumer:
            print(msg.value)

        if consumer is not None:
            consumer.close()
            
    except Exception as ex:
        logger.error('Exception: {}'.format(ex))
    return