操作场景
随着 Kafka 社区的繁荣，越来越多的用户开始使用 Kafka 来进行日志收集、大数据分析、流式数据处理等操作。而腾讯云消息队列 Ckafka 也借助了开源社区的力量，进行了如下优化：
基于 ApacheKafka 的分布式、高可扩展、高吞吐。
100%兼容 Apache KafkaAPI（0.9及0.10）。
无需部署，直接使用 Kafka 所有功能。
封装所有集群细节，无需用户运维。
对消息引擎优化，性能比社区最高提升50%。
腾讯云云函数与 Ckafka 也进行了深度联动，并推出了很多实用的功能。借助云函数和 Ckafka 触发器，可以非常方便实现 CKafka 消息转存到 COS、ES、DB 等，本文介绍使用云函数替代 Logstash，实现 Ckafka 消息落盘 ES。如下图所示： 
运行原理
云函数可以实时消费 Ckafka 中的消息。例如，做数据转存、日志清洗、实时消费等。且数据转存的功能已集成到 Ckafka 控制台，用户可以一键开启使用，降低了用户使用的复杂度。如下图所示： n
方案优势
对比使用云服务器自建 Ckafka Consumer 的方式，云函数具备以下优势：
云函数控制台支持一键开启 Ckafka 触发器，帮助用户自动创建 Consumer，并由云函数平台来维护组件的高可用。
Ckafka 触发器自身支持很多实用的配置：支持配置 offset 位置、支持配置1 - 1万消息聚合条数、支持配置1 - 1万次重试次数等。
基于云函数开发的业务逻辑，天然支持弹性伸缩，无需额外搭建和维护服务器集群等。
使用云函数和使用云服务器 CVM 自建 Logstash 对比，云函数具备以下优势：
云函数自带 Consumer 组件，可自行聚合。
云函数的模板函数已经实现了消息聚合和部分清洗能力，支持自行扩展。
云函数集群自带高可用和监控日志能力，业务上线速度更快。
云函数采用按实际使用收费，比自建集群费用更低廉，可以节省50%的费用。
前提条件
本文以广州地域为例：
需开启 Elasticsearch Service 服务。
需开启 Ckafka 服务。
操作步骤
创建云函数及 Ckafka 触发器
1. 登录 Serverless 控制台，单击左侧导航栏的函数服务。
2. 在“函数服务”上方选择期望创建函数的地域，并单击新建，进入函数创建流程。
3. 在“新建函数”页面根据以下信息选择函数模板，并单击下一步。如下图所示： n
创建方式：选择模板创建。
模糊搜索：输入“Ckafka 消息转储至 ES”，并进行搜索。本文以运行环境 Python3.6 为例。 单击模板中的查看详情，即可在弹出的“模板详情”窗口中查看相关信息，支持下载操作。
4. 在基础配置中，函数名称已经自动生成，可根据需要自行修改。按照引导配置环境变量、运行角色和私有网络，如下图所示： n
环境变量：新增如下环境变量，参考表格进行填写。如下图所示： n
key | value | 是否必填 |
ES_Address | ES 服务地址 | 是 |
ES_User | ES 用户名，默认为 elastic。 | 是 |
ES_Password | ES 用户登录密码。 | 是 |
ES_Index_KeyWord | ES 关键词索引。 | 是 |
ES_Log_IgnoreWord | 需要删除的关键词，缺省则全量写入。例如，填写 name 或 password。 | 否 |
ES_Index_TimeFormat | 按照天或者小时设置 Index，缺省则按照天建立索引。例如填写 hour。 | 否 |
运行角色：勾选“启用”，选择“配置并使用SCF模板运行角色”，将会自动创建并选择关联了 ES、Ckafka 全读写权限的 SCF 模板运行角色，或选择“使用已有角色”，在下拉列表中选择包含上述权限的已有角色。本文以“配置并使用SCF模板运行角色”为例。
私有网络：勾选“启用”，并选择与 ES 相同的 VPC。
5. 在触发器配置中，选择“自定义创建”，根据页面的参数信息进行填写。如下图所示： n
触发方式：选择 “Ckafka触发”。
Ckafka实例及 Topic：按需选择对应的 Topic。
起始位置：选择“从最开始位置开始消费”。
6. 单击完成，即可完成函数和触发器创建。
查看 ES 和函数运行日志
注意
选择函数侧边栏日志查询，即可查看函数运行日志。如下图所示： n
查看 Kibana。详情请参见 通过 Kibana 访问集群。n
扩展能力
若您需实现高级日志清洗逻辑，可在如下图所示的代码位置中修改逻辑：n
