解决临界点数据未被扫描到问题和扫描到旧数据问题

数据同步注意临界点数据不被扫描

mysql.cofig

配置

下载

bin 目录：引入mysql连接接Java jar 包

config 文件夹 mysql.conf设置

logstash配置

1、设置读取MySQL并且同步到ES的配置文件：

    input

    指定jdbc驱动jar包位置、驱动类、URL

    *设置clean_run=true，清除上次缓存的数据；

    配置执行语句。

  output

    连接ES地址:IP:port

    索引

    文档id

2、删除索引

3、logstash启动

4、验证同步结果

    使用空的搜索参数

5、执行SQL三次迭代：

    a.遗漏R12、R22等临界时间点数据

    b.造成R11等临界点数据重复，造成额外性能开销

    c.解决临界点数据遗漏和数据重复问题

• 异常一：启动报无法找到主类解决方案

找到 /bin/logstash.bat，给 %CLASSPATH% 加上双引号,原因：https://www.cnblogs.com/sbj-dawn/p/8549369.html

异常二：logstach启动成功，但是同步mysql数据出错

• 需要改下/config/mysql.conf  

characterEncoding=utf8&useSSL=false&serverTimezone=UTC&rewriteBatchedStatements=true

1、使用logstash执行如下命令时，

logstash -f ../config/mysql.conf

出现jdk包找不到的情况，参照如下地址操作，即可解决问题：

https://www.cnblogs.com/sbj-dawn/p/8549369.html

2、还有提示配置文件不是utf-8的，文件另存为改一下编码格式就好了。

input {

jdbc {

#jdbc驱动包位置

jdbc_driver_library => "/Volumes/机械/logstash-7.8.0/mysql-connector-java-5.1.49.jar"

#要使用的驱动包类型

jdbc_driver_class => "com.mysql.jdbc.Driver"

#mysql数据库连接信息

jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/mzl-pay"

#mysql登陆用户名

jdbc_user => "root"

#mysql登陆密码

jdbc_password => "rootroot"

#定时任务，多久执行一次，默认一分钟，如果自定义可以使用schedule => "* * * * *"

schedule => "* * * * *"

#清空上次的sql_last_value记录

clean_run => true

#执行的语句

statement => "select * from test_user where updatetime > :sql_last_value and updatetime < NOW() order by updatetime desc"

}

}

output {

elasticsearch {

#ES的host:port

hosts => ["http://localhost:9200"]

#放在哪个索引下

index => "user"

#表达式上面sql查询语句的id作为doc_id

document_id => "%{id}"

}

}

3.0版本，解决数据同步时数据不一致问题

同步方法，es官方提供的logstash

命令： logstash -f ../config/mysql.conf

mysql.conf:

input {

  jdbc {

jdbc_driver_library => "E:\\es\logstash-6.5.2\\mysql-connector-java-5.1.43.jar"

    jdbc_driver_class => "com.mysql.jdbc.Driver"

    jdbc_connection_string => "jdbc:mysql://localhost:3306/mp"

    jdbc_user => "root"

    jdbc_password => "ROOT"

    schedule => "* * * * *"

    clean_run => true

    statement => "select * from tb_blog  where create_time > :sql_last_value and create_time < NOW() order by create_time desc"

  }

}

output {

  elasticsearch {

    hosts => ["http://localhost:9200"]

    index => "blog"

    document_id => "%{id}"

  }

}

mysql和es数据同步，es官方提供的logstash

迁移命令： logstash -f ../config/mysql.conf

配置内容：

input {

  jdbc {

    jdbc_driver_library => "D:\\es\logstash-7.6.2\\mysql-connector-java-5.1.37-bin.jar"

    jdbc_driver_class => "com.mysql.jdbc.Driver"

    jdbc_connection_string => "jdbc:mysql://192.168.0.132:3306/estest"

    jdbc_user => "root"

    jdbc_password => "root"

    schedule => "* * * * *"

    clean_run => true

    statement => "select * from user where createtime > :sql_last_value and createtime < NOW() order by createtime desc"

  }

}

output {

  elasticsearch {

    hosts => ["http://localhost:9200"]

    index => "user"

    document_id => "%{code}"

  }

}

C:\soft\elastic\logstash-6.3.2\bin>logstash -f ../config/mysql.conf

output

config/mysql.conf

input

同步数据防止重复数据的sql

官网：https://www.elastic.co/cn/    

https://www.elastic.co/cn/products/logstash

Java技术栈目前有三种可以选择 Node Client, Transport Client, Rest API, 需要注明的是，官方已经标明NodeClient 已经过期，Transport Client 将在7.x版本开始不再支持， 最终会在7.x 统一到Rest API。目前Transport Client使用范围比较广；Rest API方式兼容性较好；除非在In-memory模式下运行单元测试，否则不推荐NodeClient

input {    

jdbc {        

# jdbc 驱动包位置

jdbc_driver_library => "F\\logstash-6.3.2 \\mysql-connector-java-5.1.37.jar"        

# 要使用的驱动包类

jdbc_driver_class => "com.mysql.jdbc.Driver"        

# mysql数据库的连接信息

jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/blog"        

# mysql 用户

jdbc_user => "root"        

# mysql 用户密码

jdbc_password => "123456"        

# 定时任务，多久执行一次查询，默认一分钟，如果想要没有延迟，可以使用"* * * * *"

schedule => "* * * * *"        

# 清空上次的sql——last_value的记录

clean_run => true        

# 要执行的sql

statement => "select * FROM t_blog WHERE update_time > :sql_last_value AND update_time < NOW() ORDER BY update_time desc"    

}

}

output {    

elasticsearch{        

# es host : host

hosts => ["127.0.0.1:9200"]        

# 索引

index => "blog"        

# _id

document_id => "%{id}"    

}

}