使用Logstash 导入log文件中数据到Mysql数据库

一、安装Logstash

Logstash从5.x版本开始必须是运行在java8的环境，所以需要检查Java版本

# 查看Java版本
java -version

具体安装方法这里不多做赘述自行考古或者百度

二、安装Logstash插件

进入Logstash根目录，我这里的目录 /usr/share/logstash

（一） 安装logstash-out-jdbc

# 安装
./bin/logstash-plugin install logstash-out-jdbc

# 卸载
./bin/logstash-plugin uninstall logstash-out-jdbc

使用 logstash-out-jdbc 该插件可以正常导入数据到Mysql中，但是效率比较低约 5-8k/s

（二）安装logstash-out-analyticdb

# 安装
./bin/logstash-plugin install logstash-out-analyticdb

# 卸载
./bin/logstash-plugin install logstash-out-analyticdb

使用logstash-out-analyticdb 效率实测比 logstash-out-jdbc 效率要高很多，具体详细介绍文档： https://help.aliyun.com/document_detail/144694.html

两个插件的使用方法都是一样的！！！

• 提示！！！： 这里如果安装比较慢的话可以耐心等待一小会，等待不了的话，百度一下切换安装源

三、安装 Mysql 连接驱动

# 获取 Mysql 驱动
wget http://central.maven.org/maven2/mysql/mysql-connector-java/5.1.49/mysql-connector-java-5.1.49.jar
# 放入指定目录，这里可以自定义，到时候在配置文件中 driver_jar_path 项填入对应绝对路径就好了

四、配置文件

新建一个logstash配置文件，名称可以自行定义，内容如下

input {
  file {
      path => "/home/log/test/test_2022-07-04.log*"
      # logstash从什么位置开始读取文件数据，默认是结束位置，也就是说 logstash 进程会以类似tail -F的形式运行。如果你是要导入原有数据，把这个设定改成"beginning"。 
      start_position => "beginning"
      # 每隔多久检查一次被监听文件状态（是否有更新），默认是1秒
      stat_interval => "2"
  }
}

filter {
   mutate {
     # 按 “,” 进行 split 切割 message
     split => ["message",","]
     add_field => {
       "time" => "%{[message][0]}"
     }
     add_field => {
       "name" => "%{[message][1]}"
     }
     add_field => {
       "age" => "%{[message][2]}"
     }

#     source => "message" 
   }

  date {
     match => ["time","yyyy-MM-dd HH:mm:ss"]
  }
}

output {

   stdout {
      codec => rubydebug
   }
  # jdbc {
  #   # 指定 Mysql 连接驱动
  #    driver_jar_path => "/usr/share/logstash/logstash-core/lib/jars/mysql-connector-java-5.1.49.jar"
  #    driver_class => "com.mysql.jdbc.Driver"
  #	  # 这里需要注意，如果你的密码中含有 [ & ] 符号，请变成转义字符 [ & ]
  #    connection_string => "jdbc:mysql://127.0.0.1:3306/test?user=root&password=test"
  #    statement => ["insert into test_20220704 (time,name,age) VALUES (?,?,?)","time","name","age"]
  # }
   analyticdb {
	  # 指定 Mysql 连接驱动
      driver_jar_path => "/usr/share/logstash/logstash-core/lib/jars/mysql-connector-java-5.1.49.jar"
      driver_class => "com.mysql.jdbc.Driver"
	  # 这里需要注意，如果你的密码中含有 [ & ] 符号，请变成转义字符 [ & ]
      connection_string => "jdbc:mysql://127.0.0.1:3306/test?user=root&password=test"
      statement => ["insert into test_20220704 (time,name,age) VALUES (?,?,?)","time","name","age"]
   }
}

五、运行实测

# 启动命令
## nohup 后台启动
## -f 指定启动配置文件
nohup /usr/share/logstash/bin/logstash -f /etc/logstash/logstash-logtomysql.conf &

同步查看运行日志信息以及Mysql表中数据量，到此就结束了！！！

这里还有一个问题，欢迎大佬指点：start_position => “beginning” ，
如果log文件实时更新，logstash是会检测到数据更新并执行导入，
但是这里会重新导入一遍当前文件中的所有数据，如果需要增量导入的需要做什么操作，
欢迎各位大佬指点!!