elasticsearch 使用 Logstash 做数据采集

2024-08-22 16:08

本文主要是介绍elasticsearch 使用 Logstash 做数据采集,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

1,下载
下载地址(根据自己需要的版本下载):
https://www.elastic.co/cn/downloads/logstash

我这里是使用的6.2.1版本,直接下载就可以了

wget https://artifacts.elastic.co/downloads/logstash/logstash-6.2.1.tar.gz

2,解压

tar -zxvf logstash-6.2.1.tar.gz

将解压后的目录移动到/usr/local/目录下

mv logstash-6.2.1 /usr/local/
cd /usr/local/logstash-6.2.1/

3,安装 logstash 所需依赖 ruby 和 rubygems(注意:需要 ruby 的版本在 1.8.7 以上)

yum install -y ruby rubygems

检查 ruby 版本

ruby -v

输出如下,表示安装成功
在这里插入图片描述
4,安装 logstash-input-jdbc

cd /usr/local/logstash-6.2.1/
./bin/logstash-plugin install --no-verify  logstash-input-jdbc

5,编写配置文件
我这里的配置文件主要是2个配置文件,mysql同步表文件(mysql.conf)和索引库映射文件(question_template.json),都放在 logstash 的 config 配置文件下
1,mysql.conf

input {stdin {}jdbc {jdbc_connection_string => "jdbc:mysql://192.168.1.1:3306/java_interview_dev?characterEncoding=utf-8&autoReconnect=true&zeroDateTimeBehavior=convertToNull&useSSL=false&serverTimezone=Asia/Shanghai"# the user we wish to excute our statement asjdbc_user => "root"jdbc_password => "123456"# the path to our downloaded jdbc driverjdbc_driver_library => "/usr/local/logstash-6.2.1/lib/mysql-connector-java-8.0.16.jar"# the name of the driver class for mysqljdbc_driver_class => "com.mysql.cj.jdbc.Driver"jdbc_paging_enabled => "true"jdbc_page_size => "50000"#要执行的sql文件#statement_filepath => "/conf/course.sql"statement => "SELECT question_id, title, answer, type_ids, DATE_FORMAT(create_time, '%Y-%m-%d %H:%i:%S') AS create_time FROM question WHERE `timestamp` > DATE_ADD(:sql_last_value,INTERVAL 8 HOUR)"#定时配置schedule => "* * * * *"record_last_run => true#记录最后采集时间点,保存到logstash_metadata文件中last_run_metadata_path => "/usr/local/logstash-6.2.1/config/logstash_metadata"}
}output {elasticsearch {#ES的ip地址和端口hosts => "localhost:9200"#hosts => ["localhost:9200"]#ES索引库名称index => "question_dev"document_id => "%{question_id}"document_type => "doc"template =>"/usr/local/logstash-6.2.1/config/question_template.json"template_name =>"question_dev"template_overwrite =>"true"}stdout {#日志输出codec => json_lines}
}

2,question_template.json

{"mappings": {"doc": {"properties": {"question_id": {"type": "integer"},"title": {"type": "text","analyzer": "ik_max_word","search_analyzer": "ik_smart"},"answer": {"type": "text","analyzer": "ik_max_word","search_analyzer": "ik_smart"},"type_ids": {"type": "text"},"create_time": {"format": "yyyy-MM-dd HH:mm:ss","type": "date"}}}},"template": "question_dev"
}

6,运行

/usr/local/logstash-6.2.1/bin/logstash -f /usr/local/logstash-6.2.1/config/mysql.conf

使 logstash 一直保持在后台运行命令:

nohup /usr/local/logstash-6.2.1/bin/logstash -f /usr/local/logstash-6.2.1/config/mysql.conf 2>&1 &

运行前:
在这里插入图片描述
索引库里面没有一条数据,
运行后:
在这里插入图片描述
运行后我们发现,logstash 会根据 mysql.conf 里面的配置项 statement 执行的sql所查询到的数据全部录入到索引库,默认的 logstash 会每分钟执行一次,可以根据配置的 schedule 定时任务修改

到这里使用 logstash 做es数据采集的过程就已经全部完成了

备注:配置不太明白的可以看我附件上传的教学视频

https://download.csdn.net/download/u012946310/11827678

备注:如果需要同时对多个数据采集并且输出到不同的索引库,参考如下配置:

input {stdin {}#dev数据库问题索引jdbc {#采集类型,避免输出时混淆,使用此类型判断输出type => "dev_question"jdbc_connection_string => "jdbc:mysql://localhost:3306/cx_blockchain_dev?characterEncoding=utf-8&autoReconnect=true&zeroDateTimeBehavior=convertToNull&useSSL=false&serverTimezone=Asia/Shanghai"# the user we wish to excute our statement asjdbc_user => "root"jdbc_password => "cx123456789cx"# the path to our downloaded jdbc driverjdbc_driver_library => "/usr/local/logstash-6.2.1/lib/mysql-connector-java-8.0.16.jar"# the name of the driver class for mysqljdbc_driver_class => "com.mysql.cj.jdbc.Driver"jdbc_paging_enabled => "true"jdbc_page_size => "50000"#要执行的sql文件#statement_filepath => "/conf/course.sql"statement => "SELECT question_id, title, `desc`, label_code, answer_count, create_user_id, DATE_FORMAT(create_time, '%Y-%m-%d %H:%i:%S') AS create_time FROM question WHERE `timestamp` > DATE_ADD(:sql_last_value,INTERVAL 8 HOUR)"#定时配置#schedule => "*/10 * * * *"schedule => "* * * * *"record_last_run => true#记录最后采集时间点,保存到dev_question_run_log文件中last_run_metadata_path => "/usr/local/logstash-6.2.1/config/es-conf/dev_question_run_log"}#test数据库问题索引jdbc {#采集类型,避免输出时混淆,使用此类型判断输出type => "test_question"jdbc_connection_string => "jdbc:mysql://localhost:3306/cx_blockchain_test?characterEncoding=utf-8&autoReconnect=true&zeroDateTimeBehavior=convertToNull&useSSL=false&serverTimezone=Asia/Shanghai"# the user we wish to excute our statement asjdbc_user => "root"jdbc_password => "cx123456789cx"# the path to our downloaded jdbc driverjdbc_driver_library => "/usr/local/logstash-6.2.1/lib/mysql-connector-java-8.0.16.jar"# the name of the driver class for mysqljdbc_driver_class => "com.mysql.cj.jdbc.Driver"jdbc_paging_enabled => "true"jdbc_page_size => "50000"#要执行的sql文件#statement_filepath => "/conf/course.sql"statement => "SELECT question_id, title, `desc`, label_code, answer_count, create_user_id, DATE_FORMAT(create_time, '%Y-%m-%d %H:%i:%S') AS create_time FROM question WHERE `timestamp` > DATE_ADD(:sql_last_value,INTERVAL 8 HOUR)"#定时配置#schedule => "*/12 * * * *"schedule => "* * * * *"record_last_run => true#记录最后采集时间点,保存到test_question_run_log文件中last_run_metadata_path => "/usr/local/logstash-6.2.1/config/es-conf/test_question_run_log"}
}output {#dev_question索引输出if[type]=="dev_question"{elasticsearch {#ES的ip地址和端口hosts => "localhost:9200"#hosts => ["localhost:9200"]#ES索引库名称index => "dev_question"document_id => "%{question_id}"document_type => "doc"template =>"/usr/local/logstash-6.2.1/config/es-conf/question_template.json"template_name =>"question"template_overwrite =>"true"}stdout {#日志输出codec => json_lines}}#test_question索引输出if[type]=="test_question"{elasticsearch {#ES的ip地址和端口hosts => "localhost:9200"#hosts => ["localhost:9200"]#ES索引库名称index => "test_question"document_id => "%{question_id}"document_type => "doc"template =>"/usr/local/logstash-6.2.1/config/es-conf/question_template.json"template_name =>"question"template_overwrite =>"true"}stdout {#日志输出codec => json_lines}}
}

上面主要新增了一个 type 字段,并且在输出的时候判断 type 字段,以此来区分采集的数据输出到不同的索引库

这篇关于elasticsearch 使用 Logstash 做数据采集的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1096745

相关文章

Java中的Cursor使用详解

《Java中的Cursor使用详解》本文介绍了Java中的Cursor接口及其在大数据集处理中的优势,包括逐行读取、分页处理、流控制、动态改变查询、并发控制和减少网络流量等,感兴趣的朋友一起看看吧... 最近看代码,有一段代码涉及到Cursor,感觉写法挺有意思的。注意是Cursor,而不是Consumer

javaScript在表单提交时获取表单数据的示例代码

《javaScript在表单提交时获取表单数据的示例代码》本文介绍了五种在JavaScript中获取表单数据的方法:使用FormData对象、手动提取表单数据、使用querySelector获取单个字... 方法 1:使用 FormData 对象FormData 是一个方便的内置对象,用于获取表单中的键值

Node.js net模块的使用示例

《Node.jsnet模块的使用示例》本文主要介绍了Node.jsnet模块的使用示例,net模块支持TCP通信,处理TCP连接和数据传输,具有一定的参考价值,感兴趣的可以了解一下... 目录简介引入 net 模块核心概念TCP (传输控制协议)Socket服务器TCP 服务器创建基本服务器服务器配置选项服

如何使用CSS3实现波浪式图片墙

《如何使用CSS3实现波浪式图片墙》:本文主要介绍了如何使用CSS3的transform属性和动画技巧实现波浪式图片墙,通过设置图片的垂直偏移量,并使用动画使其周期性地改变位置,可以创建出动态且具有波浪效果的图片墙,同时,还强调了响应式设计的重要性,以确保图片墙在不同设备上都能良好显示,详细内容请阅读本文,希望能对你有所帮助...

Rust中的注释使用解读

《Rust中的注释使用解读》本文介绍了Rust中的行注释、块注释和文档注释的使用方法,通过示例展示了如何在实际代码中应用这些注释,以提高代码的可读性和可维护性... 目录Rust 中的注释使用指南1. 行注释示例:行注释2. 块注释示例:块注释3. 文档注释示例:文档注释4. 综合示例总结Rust 中的注释

Linux使用cut进行文本提取的操作方法

《Linux使用cut进行文本提取的操作方法》Linux中的cut命令是一个命令行实用程序,用于从文件或标准输入中提取文本行的部分,本文给大家介绍了Linux使用cut进行文本提取的操作方法,文中有详... 目录简介基础语法常用选项范围选择示例用法-f:字段选择-d:分隔符-c:字符选择-b:字节选择--c

Rust中的BoxT之堆上的数据与递归类型详解

《Rust中的BoxT之堆上的数据与递归类型详解》本文介绍了Rust中的BoxT类型,包括其在堆与栈之间的内存分配,性能优势,以及如何利用BoxT来实现递归类型和处理大小未知类型,通过BoxT,Rus... 目录1. Box<T> 的基础知识1.1 堆与栈的分工1.2 性能优势2.1 递归类型的问题2.2

使用Go语言开发一个命令行文件管理工具

《使用Go语言开发一个命令行文件管理工具》这篇文章主要为大家详细介绍了如何使用Go语言开发一款命令行文件管理工具,支持批量重命名,删除,创建,移动文件,需要的小伙伴可以了解下... 目录一、工具功能一览二、核心代码解析1. 主程序结构2. 批量重命名3. 批量删除4. 创建文件/目录5. 批量移动三、如何安

springboot的调度服务与异步服务使用详解

《springboot的调度服务与异步服务使用详解》本文主要介绍了Java的ScheduledExecutorService接口和SpringBoot中如何使用调度线程池,包括核心参数、创建方式、自定... 目录1.调度服务1.1.JDK之ScheduledExecutorService1.2.spring

Java使用Tesseract-OCR实战教程

《Java使用Tesseract-OCR实战教程》本文介绍了如何在Java中使用Tesseract-OCR进行文本提取,包括Tesseract-OCR的安装、中文训练库的配置、依赖库的引入以及具体的代... 目录Java使用Tesseract-OCRTesseract-OCR安装配置中文训练库引入依赖代码实