操作场景
随着 Kafka 社区的繁荣,越来越多的用户开始使用 Kafka 来进行日志收集、大数据分析、流式数据处理等操作。而腾讯云消息队列 Ckafka 也借助了开源社区的力量,进行了如下优化:
基于 ApacheKafka 的分布式、高可扩展、高吞吐。
100%兼容 Apache KafkaAPI(0.9及0.10)。
无需部署,直接使用 Kafka 所有功能。
封装所有集群细节,无需用户运维。
对消息引擎优化,性能比社区最高提升50%。
腾讯云云函数与 Ckafka 也进行了深度联动,并推出了很多实用的功能。借助云函数和 Ckafka 触发器,可以非常方便实现 CKafka 消息转存到 COS、ES、DB 等,本文介绍使用云函数替代 Logstash,实现 Ckafka 消息落盘 ES。如下图所示: ![](https://qcloudimg.tencent-cloud.cn/image/document/ae9017a1aeb3037c7e76af6424afffba.png)
![](https://qcloudimg.tencent-cloud.cn/image/document/ae9017a1aeb3037c7e76af6424afffba.png)
运行原理
云函数可以实时消费 Ckafka 中的消息。例如,做数据转存、日志清洗、实时消费等。且数据转存的功能已集成到 Ckafka 控制台,用户可以一键开启使用,降低了用户使用的复杂度。如下图所示: n![](https://qcloudimg.tencent-cloud.cn/image/document/1bd12606fdd02829bb1c6daeeb39ead0.png)
![](https://qcloudimg.tencent-cloud.cn/image/document/1bd12606fdd02829bb1c6daeeb39ead0.png)
方案优势
对比使用云服务器自建 Ckafka Consumer 的方式,云函数具备以下优势:
云函数控制台支持一键开启 Ckafka 触发器,帮助用户自动创建 Consumer,并由云函数平台来维护组件的高可用。
Ckafka 触发器自身支持很多实用的配置:支持配置 offset 位置、支持配置1 - 1万消息聚合条数、支持配置1 - 1万次重试次数等。
基于云函数开发的业务逻辑,天然支持弹性伸缩,无需额外搭建和维护服务器集群等。
使用云函数和使用云服务器 CVM 自建 Logstash 对比,云函数具备以下优势:
云函数自带 Consumer 组件,可自行聚合。
云函数的模板函数已经实现了消息聚合和部分清洗能力,支持自行扩展。
云函数集群自带高可用和监控日志能力,业务上线速度更快。
云函数采用按实际使用收费,比自建集群费用更低廉,可以节省50%的费用。
前提条件
本文以广州地域为例:
需开启 Elasticsearch Service 服务。
需开启 Ckafka 服务。
操作步骤
创建云函数及 Ckafka 触发器
1. 登录 Serverless 控制台,单击左侧导航栏的函数服务。
2. 在“函数服务”上方选择期望创建函数的地域,并单击新建,进入函数创建流程。
3. 在“新建函数”页面根据以下信息选择函数模板,并单击下一步。如下图所示: n![](https://qcloudimg.tencent-cloud.cn/image/document/261fddc188946c77205a81326fe8f5d6.png)
![](https://qcloudimg.tencent-cloud.cn/image/document/261fddc188946c77205a81326fe8f5d6.png)
创建方式:选择模板创建。
模糊搜索:输入“Ckafka 消息转储至 ES”,并进行搜索。本文以运行环境 Python3.6 为例。 单击模板中的查看详情,即可在弹出的“模板详情”窗口中查看相关信息,支持下载操作。
4. 在基础配置中,函数名称已经自动生成,可根据需要自行修改。按照引导配置环境变量、运行角色和私有网络,如下图所示: n![](https://qcloudimg.tencent-cloud.cn/image/document/d84015fffb136854073db2f899c9ed0d.png)
![](https://qcloudimg.tencent-cloud.cn/image/document/d84015fffb136854073db2f899c9ed0d.png)
环境变量:新增如下环境变量,参考表格进行填写。如下图所示: n![](https://qcloudimg.tencent-cloud.cn/image/document/75be7f0bd89a9485166abf3cbf30cb34.png)
![](https://qcloudimg.tencent-cloud.cn/image/document/75be7f0bd89a9485166abf3cbf30cb34.png)
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![](https://qcloudimg.tencent-cloud.cn/image/document/ff76aa2a11d465110537370a5e212bfa.png)
n 主要参数信息如下,其余参数请保持默认配置:
![](https://qcloudimg.tencent-cloud.cn/image/document/ff76aa2a11d465110537370a5e212bfa.png)
触发方式:选择 “Ckafka触发”。
Ckafka实例及 Topic:按需选择对应的 Topic。
起始位置:选择“从最开始位置开始消费”。
6. 单击完成,即可完成函数和触发器创建。
查看 ES 和函数运行日志
注意
选择函数侧边栏日志查询,即可查看函数运行日志。如下图所示: n![](https://qcloudimg.tencent-cloud.cn/image/document/3f1bc15c41e6c8f075fc6493438fbfbc.png)
![](https://qcloudimg.tencent-cloud.cn/image/document/3f1bc15c41e6c8f075fc6493438fbfbc.png)
查看 Kibana。详情请参见 通过 Kibana 访问集群。n![](https://qcloudimg.tencent-cloud.cn/image/document/82349189d554d0eefa0a38272ae3f89e.png)
![](https://qcloudimg.tencent-cloud.cn/image/document/82349189d554d0eefa0a38272ae3f89e.png)
扩展能力
若您需实现高级日志清洗逻辑,可在如下图所示的代码位置中修改逻辑:n![](https://qcloudimg.tencent-cloud.cn/image/document/523afb5fcf1906dd2edb0388f7b2df34.png)
![](https://qcloudimg.tencent-cloud.cn/image/document/523afb5fcf1906dd2edb0388f7b2df34.png)