使用logstash-6.5.4同步mysql 数据到ES里面

约定不等于承诺〃 2022-12-26 10:19 170阅读 0赞

1.ES的全套下载地址 : [https://download.csdn.net/download/zhangshengqiang168/13452585][https_download.csdn.net_download_zhangshengqiang168_13452585]

2.将下载好的logstash-6.5.4解压和es放到一个目录，放到哪个目录没有关系。

![20201204162539432.png][]

3.将下载好的ysql8的驱动，放入到bin目录下，

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nc2hlbmdxaWFuZzE2OA_size_16_color_FFFFFF_t_70][]

4.配置mysql同步文件，新建mysqltoes.conf（名字没关系）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nc2hlbmdxaWFuZzE2OA_size_16_color_FFFFFF_t_70 1][]

5.编辑新建的mysqltoes.conf , 添加配置信息,(根据自己实际情况配置)

input {
     stdin { }
        jdbc {
        #填写你的mysql链接串8以后驱动必须这样写，不然后出错，这个问题我搞了好几天才解决
            jdbc_connection_string => "jdbc:mysql://localhost:3306/permission?characterEncoding=utf8&useSSL=false&serverTimezone=UTC&rewriteBatchedStatements=true"
            #链接数据库用户名称
            jdbc_user => "root"
            #链接数据库的密码
            jdbc_password => "root"
            #指定驱动的位置
            jdbc_driver_library => "C:/ES/logstash-6.5.4/bin/mysql-connector-java-8.0.19.jar"
            #最新的mysql驱动写法，写以前的驱动会报错
            jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
            jdbc_paging_enabled => "true"
            jdbc_page_size => "50000"
            #同步的表，这里也可以只想一个写了sql的文件
            statement => "SELECT * FROM member"
            #表示每分钟都同步数据
            schedule => "* * * * *"
        }
     }
     filter {
      date {
        # 有多个项的话能匹配多个不同的格式
        match => [ "logdate", "MMM dd yyyy HH:mm:ss","ISO8601" ]
        target => "fieldName1"
        timezone => "Asia/Shanghai"
      }
    }
    
     output {
         stdout {
            codec => json_lines
        }
        elasticsearch {
        #数据到es
            hosts => "localhost:9200"
            #指定索引，名字任意
            index => "gold"
            #指定类型，任意
            document_type => "doc"
            document_id => "%{id}"
        }
    }

6.多表配置,(多张表同步,可以配置多个jdbc\{\} )

input {
        stdin {
        
        }
        jdbc {
    	  type => "talk"
           # mysql 数据库链接,mybatis为数据库名
          jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/talk?useUnicode=true&useSSL=false&characterEncoding=utf8&serverTimezone=UTC&allowPublicKeyRetrieval=true"
          # 用户名和密码
          jdbc_user => "root"
          jdbc_password => "root"
          # 驱动 修改为mysql连接包位置
          jdbc_driver_library => "D:/tool/elasticsearch/logstash-6.5.4/bin/bin/mysql-connector-java-8.0.19.jar"
          # 驱动类名
          jdbc_driver_class => "com.mysql.jdbc.Driver"
          jdbc_paging_enabled => "true"
          jdbc_page_size => "50000"
          # 执行的sql 文件路径+名称 
          statement_filepath => "D:/tool/elasticsearch/logstash-6.5.4/bin/talkselect.sql"
          # 设置监听间隔  各字段含义（由左至右）分、时、天、月、年，全部为*默认含义为每分钟都更新
          schedule => "* * * * *"
        }
    	
    	 jdbc {
    	  type => "video"
           # mysql 数据库链接,mybatis为数据库名
          jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/video?useUnicode=true&useSSL=false&characterEncoding=utf8&serverTimezone=UTC&allowPublicKeyRetrieval=true"
          # 用户名和密码
          jdbc_user => "root"
          jdbc_password => "root"
          # 驱动 修改为mysql连接包位置
          jdbc_driver_library => "D:/tool/elasticsearch/logstash-6.5.4/bin/bin/mysql-connector-java-8.0.19.jar"
          # 驱动类名
          jdbc_driver_class => "com.mysql.jdbc.Driver"
          jdbc_paging_enabled => "true"
          jdbc_page_size => "50000"
          # 执行的sql 文件路径+名称 
          statement_filepath => "D:/tool/elasticsearch/logstash-6.5.4/bin/videoselect.sql"
          # 设置监听间隔  各字段含义（由左至右）分、时、天、月、年，全部为*默认含义为每分钟都更新
          schedule => "* * * * *"
        }
    	
    		
    	
    }
    
    filter {
      mutate {
            remove_field => ["@timestamp","@version","create_time","update_time"]
    
      }
    }
    
    output {
    	if[type]=="talk"{
    	    elasticsearch {
    			# ES的IP地址及端口
    			hosts => ["localhost:9200"]
    			# 索引名称 可自定义
    			index => "talkindex"
    			# 需要关联的数据库中有有一个id字段，对应类型中的id
    			document_id => "%{id}"
    			document_type => "%{type}"
    		}
    	}
    	
    	if[type]=="video"{
    	    elasticsearch {
    			# ES的IP地址及端口
    			hosts => ["localhost:9200"]
    			# 索引名称 可自定义
    			index => "videoindex"
    			# 需要关联的数据库中有有一个id字段，对应类型中的id
    			document_id => "%{id}"
    			document_type => "%{type}"
    		}
    	}
    	
    
        stdout {
            # JSON格式输出
            codec => json_lines
        }
    }

7.进入 logstash的bin目录执行 logstash -f mysqltoes.conf 看到如下信息表述数据同步成功，下面截图的数据需要根据您同步的数据觉定。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nc2hlbmdxaWFuZzE2OA_size_16_color_FFFFFF_t_70 2][]

8.在kibnan的客户端里面查询Dev Tools里面查询  ,如下图,我配的索引为gold

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nc2hlbmdxaWFuZzE2OA_size_16_color_FFFFFF_t_70 3][]

[https_download.csdn.net_download_zhangshengqiang168_13452585]: https://download.csdn.net/download/zhangshengqiang168/13452585
[20201204162539432.png]: /images/20221120/990825498b494fa39987f34da621743d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nc2hlbmdxaWFuZzE2OA_size_16_color_FFFFFF_t_70]: /images/20221120/f4602eeec7c34722882a03d5e52ecd69.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nc2hlbmdxaWFuZzE2OA_size_16_color_FFFFFF_t_70 1]: /images/20221120/d7ec0f3953254369a2267c119f78931d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nc2hlbmdxaWFuZzE2OA_size_16_color_FFFFFF_t_70 2]: /images/20221120/cef817a9f2134762b21d5b0cd59a3d6a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nc2hlbmdxaWFuZzE2OA_size_16_color_FFFFFF_t_70 3]: /images/20221120/19a4034794344ca7bab86353f090bed9.png