ElasticSearch | 用 JDBC 插件导入数据到 ElasticSearch 中

需求

将数据库中的数据同步到 ES，借助 ES 的全文搜索，提高搜索的速度。

需要把新增用户信息同步到 ElasticSearch 中；
用户信息被 Update 后，需要能被更新到 ElasticSearch 中；
支持增量更新；
用户注销后，不能被 ES 搜索到；

JDBC Input Plugin

JDBC Input Plugin 可以将数据从数据库读到 Logstash。

需要自己提供所需的 JDBC Driver；
JDBC Input Plugin 支持定时任务 Scheduling，其语法来自 Rufus-scheduler，其扩展了 Cron，使用 Cron 的语法可以完成任务的触发；
JDBC Input Plugin 支持通过 Tracking_column / sql_last_value 的方式记录 State，最终实现增量的更新；

JDBC Input Plugin | 举个栗子

把驱动拷贝到 Logstash 的目录下

/home/lixinlei/application/logstash-7.7.0/logstash-core/lib/jars/mysql-connector-java-8.0.11.jar

准备好 JDBC Input Plugin 需要的 .yaml 文件

input {
  jdbc {
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://localhost:3306/db_example?useSSL=false"
    jdbc_user => root
    jdbc_password => Jiangdi_2018
    #启用增量更新，如果为true，则需要指定 tracking_column
    use_column_value => true
    #指定追踪的字段，
    tracking_column => "last_updated"
    #追踪字段的类型，目前只有数字（numeric）和时间类型（timestamp），默认是数字类型
    tracking_column_type => "numeric"
    #记录最后一次运行的结果
    record_last_run => true
    #最后一次运行的结果的保存位置，在 Logstash 的工作目录下
    last_run_metadata_path => "jdbc-position.txt"
    # 触发时执行的 SQL，sql_last_value 就是保存在 jdbc-position.txt 中的
    statement => "SELECT * FROM user where last_updated >:sql_last_value;"
    # Cron 语法，每秒做一次触发
    schedule => " * * * * * *"
  }
}
output {
  elasticsearch {
    document_id => "%{id}"
    document_type => "_doc"
    index => "users"
    hosts => ["http://localhost:9200"]
  }
  stdout{
    codec => rubydebug
  }
}

准备 Springboot 的程序

这个程序的作用主要是对 MySQL 中的一张表做增删改查；

用 mysql-demo.yaml 启动 Logstash

bin/logstash -f mysql-demo.conf

通过 Springboot 程序向 MySQL 中写入数据

curl localhost:8080/demo/add -d name=Mike -d email=mike@xyz.com -d tags=Elasticsearch,IntelliJ
curl localhost:8080/demo/add -d name=Jack -d email=jack@xyz.com -d tags=Mysql,IntelliJ
curl localhost:8080/demo/add -d name=Bob -d email=bob@xyz.com -d tags=Mysql,IntelliJ

Logstash 读到新数据，并写进 ElasticSearch；
通过 Kibana 可以查询到数据 GET users/_search；

更新 MySQL 中的数据

curl -X PUT localhost:8080/demo/update -d id=3 -d name=Bob2 -d email=bob2@xyz.com -d tags=Mysql,IntelliJ

Logstash 可以读到更新，完了把更新也写到 ElasticSearch 中；

删除 MySQL 中的用户

curl -X DELETE localhost:8080/demo/delete -d id=3

Logstash 可以读到更新，完了把更新也写到 ElasticSearch 中；

通过给索引 users 设置 alias，使得查询只查没被删除的用户

通过 view_users 查询用户，就只能查到没有被删除的用户；

POST /_aliases
{
  "actions": [
    {
      "add": {
        "index": "users",
        "alias": "view_users",
        "filter" : { 
          "term" : { 
            "is_deleted" : false 
          } 
        }
      }
    }
  ]
}

POST view_users/_search
{

}

最后编辑于：2020.06.11 14:05:25