ElasticSearch全量或增量构建索引

10次阅读

共计 3753 个字符，预计需要花费 10 分钟才能阅读完成。

ElasticSearch 集群环境搭建好当前，首次须要全量地从关系型数据库中将指标待索引数据写入到 ElasticSearch 搜索引擎中，以下咱们将用到 logstash 的插件 logstash-input-jdbc 来全量同步数据。

参考 ElasticSearch 本地集群搭建

Logstash 好比一个数据管道，通常一端连贯着关系型数据库来往管道中输出数据，一端连贯着 ElasticSearch 将管道中的数据输入。

返回官网下载 Logstash，须要与 ElasticSearch 版本统一，兼容性更好。
将下载的 Logstash 压缩包上传到服务器上，示例将 Logstash 装置在了与 ElasticSearch 同一台服务器上。

# 进入 logstash 压缩包所在目录
cd /usr/local

#加压 logstash 压缩包
unzip logstash-7.8.0.zip

装置 logstash 的插件 logstash-input-jdbc

# 进入 logstash 装置目录的 bin 目录下
cd /usr/local/logstash-7.8.0/bin

#装置插件，留神 logstash 依赖 jdk，装置插件前请确保装置了 jdk
#以后 logstash-7.8.0 版本曾经默认装置了 logstash-integration-jdbc，而此插件曾经蕴含 logstash-input-jdbc
./logstash-plugin install logstash-input-jdbc

#查看 logstash 所有已装置的插件
./logstash-plugin list

logstash 须要与 mysql 通信并获取数据，于是须要筹备与 mysql 通信的驱动程序，须要筹备拉取数据所须要执行的 sql 语句，还须要 logstash 的输出、输入端点。

cd /usr/local/logstash-7.8.0
mkdir mysql
cd mysql
touch jdbc.conf
touch jdbc.sql

mysql 驱动程序，返回 maven 地方仓库获取
获取数据的 sql 脚本（jdbc.sql）

SELECT
    a.id,
    a.NAME,
    a.tags,
    concat(a.latitude, ',', a.longitude) AS location,
    a.remark_score,
    a.price_per_man,
    a.category_id,
    b.NAME AS category_name,
    a.seller_id,
    c.remark_score AS seller_remark_score,
    c.disabled_flag AS seller_disabled_flag 
FROM
    shop a
    INNER JOIN category b ON a.category_id = b.id
    INNER JOIN seller c ON c.id = a.seller_id

logstash 输出、输入配置

input {
    jdbc {
      # mysql 数据库链接,dianping 为数据库名
      jdbc_connection_string => "jdbc:mysql://192.168.15.150:3306/dianping"
      # 用户名和明码
      jdbc_user => "root"
      jdbc_password => "123456"
      # 驱动
      jdbc_driver_library => "/usr/local/logstash-7.8.0/mysql/mysql-connector-java-5.1.47.jar"
      # 驱动类名
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      jdbc_paging_enabled => "true"
      jdbc_page_size => "50000"
    # 执行的 sql 文件门路 + 名称
      statement_filepath => "/usr/local/logstash-7.8.0/mysql/jdbc.sql"
      # 设置监听距离  各字段含意（由左至右）分、时、天、月、年，全副为 * 默认含意为每分钟都更新
      schedule => "* * * * *"
    }
}


output {
    elasticsearch {
      # ES 的 IP 地址及端口
        hosts => ["192.168.15.151:9200"]
      # 索引名称
        index => "shop"
        document_type => "_doc"
      # 自增 ID 须要关联的数据库中有有一个 id 字段，对应索引的 id 号
        document_id => "%{id}"
    }
    stdout {
     # JSON 格局输入
        codec => json_lines
    }
}

# 进入 logstash 装置目录
cd /usr/local/logstash-7.8.0

#后盾过程形式启动 logstash
nohup ./bin/logstash -f mysql/jdbc.conf &

#查看 logstash 日志
tail -f -n 100 ./nohup.out

阐明：从以上日志中看出 logstash 失常启动，并且每分钟会进行全量同步数据，日志中也打印出了执行的 sql，首先会查个总数，为了后续的分页获取数据。

其实在以上 Logstash 管道构建全量索引的根底上，稍微批改即可实现基于时间轴的增量构建索引。仅仅须要批改 jdbc.conf、新增一个记录时间的文件，批改 jdbc.sql 文件退出工夫条件即可。

SELECT
    a.id,
    a.NAME,
    a.tags,
    concat(a.latitude, ',', a.longitude) AS location,
    a.remark_score,
    a.price_per_man,
    a.category_id,
    b.NAME AS category_name,
    a.seller_id,
    c.remark_score AS seller_remark_score,
    c.disabled_flag AS seller_disabled_flag 
FROM
    shop a
    INNER JOIN category b ON a.category_id = b.id
    INNER JOIN seller c ON c.id = a.seller_id 
WHERE
    a.update_at > : sql_last_value 
    OR b.update_at > : sql_last_value 
    OR c.update_at > : sql_last_value

阐明：退出了工夫条件，管制抓取工夫窗口内的数据，增量更新。

input {
    jdbc {
      #设置时区 timezone
      jdbc_default_timezone => "Asia/Shanghai"
      # mysql 数据库链接,dianping 为数据库名
      jdbc_connection_string => "jdbc:mysql://192.168.15.150:3306/dianping"
      # 用户名和明码
      jdbc_user => "root"
      jdbc_password => "910625"
      # 驱动
      jdbc_driver_library => "/usr/local/logstash-7.8.0/mysql/mysql-connector-java-5.1.47.jar"
      # 驱动类名
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      jdbc_paging_enabled => "true"
      jdbc_page_size => "50000"
      #记录上次运行元数据文件门路，如记录增量同步的工夫窗口
      last_run_metadata_path => "/usr/local/logstash-7.8.0/mysql/last_value_meta"
    # 执行的 sql 文件门路 + 名称
      statement_filepath => "/usr/local/logstash-7.8.0/mysql/jdbc.sql"
      # 设置监听距离  各字段含意（由左至右）分、时、天、月、年，全副为 * 默认含意为每分钟都更新
      schedule => "* * * * *"
    }
}

output {
    elasticsearch {
      # ES 的 IP 地址及端口
        hosts => ["192.168.15.151:9200"]
      # 索引名称
        index => "shop"
                document_type => "_doc"
      # 自增 ID 须要关联的数据库中有有一个 id 字段，对应索引的 id 号
        document_id => "%{id}"
    }
    stdout {
     # JSON 格局输入
        codec => json_lines
    }
}

阐明：减少了时区的配置，以及指定了记录上次运行元数据文件门路，如记录增量同步的工夫窗口。

2020-07-12 00:00:00

阐明：记录上次执行同步工作时的工夫，下次工作会持续更新这个工夫。

这种以工夫窗口的形式进行数据同步，始终防止不了同步提早的问题，然而有时候业务上也是可能容许的，增量同步的工夫窗口须要更具业务来定。工夫窗口太短，数据更新太多，会呈现一次增量同步工作还没实现，下一次增量工作又开始了。

正文完

elasticsearch

发表至： elasticsearch

2020-07-12

0

关于elasticsearch:Elasticsearch集群脑裂现象探讨

关于elasticsearch:开源搜索引擎排名第一Elasticsearch是如何做到的

关于elasticsearch:mac下使用-docker-搭建单机机器集群

关于elasticsearch:elasticsearch进阶4-手动触发滚动

Python函数装饰器指南

ElasticSearch全量或增量构建索引

一、全量构建索引

1、Logstash 管道构建全量索引

下载 Logstash

装置 Logstash

配置 logstash

启动 logstash

二、增量构建索引

1、Logstash 管道构建增量索引

获取数据的 sql 脚本（jdbc.sql）

logstash 输出、输入配置

新增 last_value_meta 文件，并写入一个初始工夫

Logstash 管道构建增量索引的毛病

站内搜索