当Elasticsearch遇见Kafka

欢迎大家前往腾讯云+社区,获取更多腾讯海量技术实践干货哦~

本文由michelmu发表于云+社区专栏

Elasticsearch作为当前主流的全文检索引擎,除了强大的全文检索能力和高扩展性之外,对多种数据源的兼容能力也是其成功的秘诀之一。而Elasticsearch强大的数据源兼容能力,主要来源于其核心组件之一的Logstash, Logstash通过插件的形式实现了对多种数据源的输入和输出。Kafka是一种高吞吐量的分布式发布订阅消息系统,是一种常见的数据源,也是Logstash支持的众多输入输出源的其中一个。本文将从实践的角度,研究使用Logstash Kafka Input插件实现将Kafka中数据导入到Elasticsearch的过程。

使用Logstash Kafka插件连接Kafka和Elasticsearch

1 Logstash Kafka input插件简介

Logstash Kafka Input插件使用Kafka API从Kafka topic中读取数据信息,使用时需要注意Kafka的版本及对应的插件版本是否一致。该插件支持通过SSL和Kerveros SASL方式连接Kafka。另外该插件提供了group管理,并使用默认的offset管理策略来操作Kafka topic。

Logstash默认情况下会使用一个单独的group来订阅Kafka消息,每个Logstash Kafka Consumer会使用多个线程来增加吞吐量。当然也可以多个Logstash实例使用同一个group_id,来均衡负载。另外建议把Consumer的个数设置为Kafka分区的大小,以提供更好的性能。

2 测试环境准备

2.1 创建Elasticsearch集群

为了简化搭建过程,本文使用了腾讯云Elasticsearch service。腾讯云Elasticsearch service不仅可以实现Elasticsearch集群的快速搭建,还提供了内置Kibana,集群监控,专用主节点,Ik分词插件等功能,极大的简化了Elasticsearch集群的创建和管理工作。

2.2 创建Kafka服务

Kafka服务的搭建采用腾讯云CKafka来完成。与Elasticsearch Service一样,腾讯云CKafka可以实现Kafka服务的快速创建,100%兼容开源Kafka API(0.9版本)。

2.3 服务器

除了准备Elasticsearch和Kafka,另外还需要准备一台服务器,用于运行Logstash以连接Elasticsearch和Kafka。本文采用腾讯云CVM服务器

2.4 注意事项

1) 需要将Elasticsearch、Kafka和服务器创建在同一个网络下,以便实现网络互通。由于本文采用的是腾讯云相关的技术服务,因此只需要将Elasticsearch service,CKafka和CVM创建在同一个私有网路(VPC)下即可。

2) 注意获取Elasticsearch serivce,CKafka和CVM的内网地址和端口,以便后续服务使用

本次测试中:

服务 ip port
Elasticsearch service 192.168.0.8 9200
Ckafka 192.168.13.10 9092
CVM 192.168.0.13 -

3 使用Logstash连接Elasticsearch和Kafka

3.1 Kafka准备

可以参考[CKafka 使用入门]

按照上面的教程

1) 创建名为kafka_es_test的topic

2) 安装JDK

3) 安装Kafka工具包

4) 创建producer和consumer验证kafka功能

3.2 安装Logstash

Logstash的安装和使用可以参考[一文快速上手Logstash]

3.3 配置Logstash Kafka input插件

创建kafka_test_pipeline.conf文件内容如下:

input{
        kafka{
                bootstrap_servers=>"192.168.13.10:9092"
                topics=>["kafka_es_test"]
                group_id=>"logstash_kafka_test"
        }
}
output{
        elasticsearch{
                hosts=>["192.168.0.8:9200"]
        }
}

其中定义了一个kafka的input和一个elasticsearch的output

对于Kafka input插件上述三个参数为必填参数,除此之外还有一些对插件行为进行调整的一些参数如:

auto_commit_interval_ms 用于设置Consumer提交offset给Kafka的时间间隔

consumer_threads 用于设置Consumer的线程数,默认为1,实际中应设置与Kafka Topic分区数一致

fetch_max_wait_ms 用于指定Consumer等待一个fetch请求达到fetch_min_bytes的最长时间

fetch_min_bytes 用于指定Consumer fetch请求应返回的最小数据量

topics_pattern 用于通过正则订阅符合某一规则的一组topic

更多参数参考:[Kafka Input Configuration Options]

3.4 启动Logstash

以下操作在Logstash根目录中进行

1) 验证配置

./bin/logstash -f kafka_test_pipeline.conf --config.test_and_exit

如有错误,根据提示修改配置文件。若配置正确会得到如下结果

Sending Logstash‘s logs to /root/logstash-5.6.13/logs which is now configured via log4j2.properties
[2018-11-11T15:24:01,598][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"netflow", :directory=>"/root/logstash-5.6.13/modules/netflow/configuration"}
[2018-11-11T15:24:01,603][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"fb_apache", :directory=>"/root/logstash-5.6.13/modules/fb_apache/configuration"}
Configuration OK
[2018-11-11T15:24:01,746][INFO ][logstash.runner          ] Using config.test_and_exit mode. Config Validation Result: OK. Exiting Logstash

2) 启动Logstash

./bin/logstash -f kafka_test_pipeline.conf --config.reload.automatic

观察日志是否有错误提示,并及时处理

3.4 启动Kafka Producer

以下操作在Kafka工具包根目录下进行

./bin/kafka-console-producer.sh --broker-list 192.168.13.10:9092 --topic kafka_es_test

写入测试数据

This is a message

3.5 Kibana验证结果

登录Elasticsearch对应Kibana, 在Dev Tools中进行如下操作

1) 查看索引

GET _cat/indices

可以看到一个名为logstash-xxx.xx.xx的索引被创建成功

green open .kibana             QUw45tN0SHqeHbF9-QVU6A 1 1 1 0 5.5kb 2.7kb
green open logstash-2018.11.11 DejRdNJVQ1e1MwbyJjJjLw 5 1 1 0 8.7kb 4.3kb

2) 查看写入的数据

GET logstash-2018.11.11/_search

可以看到数据已经被成功写入

{
  "took": 0,
  "timed_out": false,
  "_shards": {
    "total": 5,
    "successful": 5,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": 1,
    "max_score": 1,
    "hits": [
      {
        "_index": "logstash-2018.11.11",
        "_type": "logs",
        "_id": "AWcBsEegMu-Dkjm1ap3H",
        "_score": 1,
        "_source": {
          "message": "This is a message",
          "@version": "1",
          "@timestamp": "2018-11-11T07:33:09.079Z"
        }
      }
    ]
  }
}

4 总结

Logstash作为Elastic Stack中数据采集和处理的核心组件,为Elasticsearch提供了强大的数据源兼容能力。从测试过程可以看出,使用Logstash实现kafka和Elaticsearch的连接过程相当简单方便。另外Logstash的数据处理功能,也使得采用该架构的系统对数据映射和处理有天然的优势。

然而,使用Logstash实现Kafka和Elasticsearch的连接,并不是连接Kafka和Elasticsearch的唯一方案,另一种常见的方案是使用Kafka Connect, 可以参考“当Elasticsearch遇见Kafka--Kafka Connect

相关阅读
【每日课程推荐】机器学习实战!快速入门在线广告业务及CTR相应知识

此文已由作者授权腾讯云+社区发布,更多原文请点击

搜索关注公众号「云加社区」,第一时间获取技术干货,关注后回复1024 送你一份技术课程大礼包!

海量技术实践经验,尽在云加社区

原文地址:http://blog.51cto.com/13957478/2315995

时间: 2024-09-30 10:42:44

当Elasticsearch遇见Kafka的相关文章

Elasticsearch 与 Kafka 整合剖析

1.概述 目前,随着大数据的浪潮,Kafka 被越来越多的企业所认可,如今的Kafka已发展到0.10.x,其优秀的特性也带给我们解决实际业务的方案.对于数据分流来说,既可以分流到离线存储平台(HDFS),离线计算平台(Hive仓库),也可以分流实时流水计算(Storm,Spark)等,同样也可以分流到海量数据查询(HBase),或是及时查询(ElasticSearch).而今天笔者给大家分享的就是Kafka 分流数据到 ElasticSearch. 2.内容 我们知道,ElasticSearc

lagstash + elasticsearch + kibana 3 + kafka 日志管理系统部署 02

因公司数据安全和分析的需要,故调研了一下 GlusterFS + lagstash + elasticsearch + kibana 3 + redis 整合在一起的日志管理应用: 安装,配置过程,使用情况等续 一,glusterfs分布式文件系统部署: 说明: 公司想做网站业务日志及系统日志统一收集和管理,经过对mfs, fastdfs 等分布式文件系统的调研,最后选择了 glusterfs,因为Gluster具有高扩展性.高性能.高可用性.可横向扩展的弹性特点,无元数据服务器设计使glust

beam与kafka和elasticSearch示例 在flink平台运行

示例实现beam用java编程,监听kafka的testmsg主题,然后将收取到的单词,按5秒做一次统计.结果输出到outputmessage 的kafka主题,同时同步到elasticSearch. kafka需要运行 启动:cd /root/kafuka/kafka_2.12-0.11.0.0 nohup bin/zookeeper-server-start.sh config/zookeeper.properties & nohup bin/kafka-server-start.sh co

JavaWeb项目架构之Kafka分布式日志队列

架构.分布式.日志队列,标题自己都看着唬人,其实就是一个日志收集的功能,只不过中间加了一个Kafka做消息队列罢了. kafka介绍 Kafka是由Apache软件基金会开发的一个开源流处理平台,由Scala和Java编写.Kafka是一种高吞吐量的分布式发布订阅消息系统,它可以处理消费者规模的网站中的所有动作流数据. 这种动作(网页浏览,搜索和其他用户的行动)是在现代网络上的许多社会功能的一个关键因素. 这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决. 特性 Kafka是一种高吞

Kafka session.timeout.ms heartbeat.interval.ms参数的区别以及对数据存储的一些思考

Kafka session.timeout.ms heartbeat.interval.ms参数的区别以及对数据存储的一些思考 在计算机世界中经常需要与数据打交道,这也是我们戏称CURD工程师的原因之一.写了两年代码,接触了不少存储系统,Redis.MySQL.Kafka.Elasticsearch-慢慢地发现背后的一些公共的设计思想总是那么似曾相识,再深究一下,就会发现一些隐藏在这些系统背后的数学理论. 生活中产生的大量数据需要交由计算机来处理,根据处理方式的不同分为OLTP和OLAP两大类应

Kubernetes 1.5集成heapster

Heapster是kubernetes集群监控工具.在1.2的时候,kubernetes的监控需要在node节点上运行cAdvisor作为agent收集本机和容器的资源数据,包括cpu.内存.网络.文件系统等.在新版的kubernetes中,cAdvisor被集成到kubelet中.通过netstat可以查看到kubelet新开了一个4194的端口,这就是cAdvisor监听的端口,现在我们然后可以通过http://<node-ip>:4194的方式访问到cAdvisor.Heapster就是

内置自研容器引擎Newbon,精灵云EcOS更新至2.1版

情人节那天,我们正式对外展现了精灵云自研容器调度引擎Newbon的真实面目(详情可参见微信).之所以选择情人节那天发布是希望接受过精灵云公有云超50万次容器调度的验证和考验的Newbon可以成为最懂广大Docker使用者的"最佳搭档". 而同样内置了Newbon的Ghostcloud精灵云企业级容器云平台EcOS(Enterprise Container Operation System)也在情人节后第三天正式更新并上线至2.1版.新版本的EcOS做了以下5点重要功能更新: 新增应用S

我关注的一周技术动态 2015.10.25

分布式系统实践 1. ScyllaDB:用 C++ 重写后的 Cassandra ,性能提高了十倍 http://blog.jobbole.com/93027/?hmsr=toutiao.io&utm_medium=toutiao.io&utm_source=toutiao.io 要点: 一直非常不喜欢hadoop系列对JVM的重度依赖, 可能是我不熟悉java的原因吧, 总感觉JVM背着我们做了很多不可见的工作, 心里不踏实. ScyllaDB宣称比Cassandra性能提高十倍, 肯定

【第五年-创业路】

[大公司] https://yq.aliyun.com/articles/71294?spm=5176.100239.blogcont73953.65.wn6WlO   微博红包技术最佳实践:架构及运维 https://yc.163yun.com/?id=ycdh&tag=ycdh_hdpt_cnb    云论坛 http://edu.csdn.net/huiyiCourse/detail/419   去哪儿网前端统一构建解决方案 http://geek.csdn.net/news/detail