摘要：配置文件其中有兩個(gè)關(guān)鍵的配置和。啟動(dòng)如上即為正常運(yùn)行。因?yàn)槲以趩?dòng)后一直報(bào)錯(cuò)，，各種嘗試最后報(bào)錯(cuò)依然存在，只好換用部署了。安裝部署安裝和插件獲取驅(qū)動(dòng)下載配置配置使用時(shí)自行把下面注釋去掉。

Author : Ali0th

Date : 20190514

最近用go語(yǔ)言寫了個(gè)爬蟲(chóng)，爬了幾百萬(wàn)條數(shù)據(jù)，存在 mysql 里，數(shù)據(jù)量較大，一個(gè)表就一兩G的程度(mysql表一般不要超過(guò)2G)。

使用查詢語(yǔ)句非常之慢，而且我要經(jīng)常使用到"LIKE",一條語(yǔ)句返回結(jié)果耗時(shí) 10~30 秒，這可不行。所以決定把數(shù)據(jù)遷移到 ES。真的是踩坑了，本來(lái)只想簡(jiǎn)單搞搞，結(jié)果搞了好多天。

本文章介紹兩種遷移方式：go-mysql-elasticsearch 和 logstash。以下是相關(guān)實(shí)踐過(guò)程。

一開(kāi)始發(fā)現(xiàn)了 go-mysql-elasticsearch 這個(gè)工具，也是go寫的，不過(guò)對(duì)ES的支持版本比較舊。于是我就把ES換成了舊版本。（新的ES版本就用logstash方式吧）

# 主要組件
mysql            : 5.5.3
elasticsearch    : 5.6.16
go-mysql-elasticsearch
# 環(huán)境
windows 10
go               : 1.11.2

同時(shí)也會(huì)用到 win 下 git 的命令行模式（可以運(yùn)行 bash 命令）

go-mysql-elasticsearch對(duì) mysql 和 ES 的要求為：

MySQL supported version < 8.0

ES supported version < 6.0

ES舊版本下載

kibana 5.6.16下載

mysql 和 elasticsearch 安裝很簡(jiǎn)單，這里就不說(shuō)明了。這里主要說(shuō) go-mysql-elasticsearch 的安裝過(guò)程。

項(xiàng)目地址：go-mysql-elasticsearch

go get github.com/siddontang/go-mysql-elasticsearch

cd到目錄下，我的 go mod 不好用，所以我直接使用 go get 安裝。

go get github.com/juju/errors
go get github.com/pingcap/check
go get github.com/siddontang/go/sync2
go get github.com/siddontang/go-mysql
go get github.com/pingcap/errors
go get github.com/shopspring/decimal
go get github.com/siddontang/go-log/log

修改代碼中的一處錯(cuò)誤。（應(yīng)該是版本原因?qū)е碌模?/p>

文件：river/river.go

編譯：

go build -o bin/go-mysql-elasticsearch ./cmd/go-mysql-elasticsearch

無(wú)報(bào)錯(cuò)，并看到有 bin/go-mysql-elasticsearch 文件即成功。

文件：my.ini

添加以下配置并重啟。

# Binary Logging
server-id=1
log_bin = mysql-bin
binlog_format = ROW

文件：etc/river.toml

其中有兩個(gè)關(guān)鍵的配置Source和rule。

[[rule]]
schema = "mysql_es"   # Mysql數(shù)據(jù)庫(kù)名
table = "test_table"  # Mysql表名
index = "test_index"  # ES中index名
type = "doc"          # 文檔類型

這一部分看官方示例就很清楚了 river.toml 和 Elasticsearch最佳實(shí)踐從Mysql到Elasticsearch

etc/river.toml 中有一處對(duì)于mysqldump的配置，只要把 mysql 的 bin 目錄加到環(huán)境變量就可以了。但我這里還是找不到mysqldump，所以我就直接到mysqldump.exe拷貝到go-mysql-elasticsearch目錄下就可以了。

./bin/go-mysql-elasticsearch -config=./etc/river.toml

如上即為正常運(yùn)行。

# 查詢各索引情況
curl -X GET "localhost:9200/_cat/indices");

# 查詢數(shù)量
curl -X GET "localhost:9200/_cat/count");
# 查詢某index的文檔數(shù)量
curl -X GET "localhost:9200/_cat/count/index_name");

# 查詢前1000條數(shù)據(jù)
curl -X GET "localhost:9200/test/_search");

ES : 7.0.1
Kibana : 7.0.1
Logstash : 7.0.1

我的 ES/Kibana 在windows 下，logstash 在centos下。

因?yàn)槲以?windows 啟動(dòng) logstash 后一直報(bào)錯(cuò)，[2019-05-15T11:55:00,183][ERROR][logstash.inputs.jdbc ] Failed to load C:/setup/mysql-connector-java-5.1.47/mysql-connector-java-5.1.47-bin.jar ，各種嘗試最后報(bào)錯(cuò)依然存在，只好換用centos部署 logstash了。

0.2.1. 安裝部署

安裝 jdbc 和 elasticsearch 插件

./bin/logstash-plugin install logstash-input-jdbc
./bin/logstash-plugin install logstash-output-elasticsearch

獲取 jdbc mysql 驅(qū)動(dòng)

下載

配置jdbc.conf,使用時(shí)自行把下面注釋去掉。

input {
    stdin {
    }
    jdbc {
      # mysql相關(guān)jdbc配置
      jdbc_connection_string => "jdbc:mysql://192.168.31.134:3306/test"
      jdbc_user => "root"
      jdbc_password => "root123"

      # jdbc連接mysql驅(qū)動(dòng)的文件目錄，可去官網(wǎng)下載:https://dev.mysql.com/downloads/connector/j/
      jdbc_driver_library => "/home/mt/Desktop/mysql-connector-java-5.1.47/mysql-connector-java-5.1.47-bin.jar"
      # the name of the driver class for mysql
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      jdbc_paging_enabled => "true"
      jdbc_page_size => "50000"

      # mysql文件, 也可以直接寫SQL語(yǔ)句在此處，如下：
      # statement => "SELECT * from Table_test;"
      # statement_filepath => "C:/setup/logstash-7.0.1/config/myconfig/jdbc.sql"
      statement => "SELECT * FROM table WHERE id >= :sql_last_value"

      # 這里類似crontab,可以定制定時(shí)操作，比如每10分鐘執(zhí)行一次同步(分 時(shí) 天 月 年)
      schedule => "*/1 * * * *"
      type => "jdbc"

      # 是否記錄上次執(zhí)行結(jié)果, 如果為真,將會(huì)把上次執(zhí)行到的 tracking_column 字段的值記錄下來(lái),保存到 last_run_metadata_path 指定的文件中
      record_last_run => "true"

      # 是否需要記錄某個(gè)column 的值,如果record_last_run為真,可以自定義我們需要 track 的 column 名稱，此時(shí)該參數(shù)就要為 true. 否則默認(rèn) track 的是 timestamp 的值.
      use_column_value => "true"

      # 如果 use_column_value 為真,需配置此參數(shù). track 的數(shù)據(jù)庫(kù) column 名,該 column 必須是遞增的. 一般是mysql主鍵
      tracking_column => "id"

      last_run_metadata_path => "/home/mt/Desktop/logstash-7.0.1/myconf/last_id"

      # 是否清除 last_run_metadata_path 的記錄,如果為真那么每次都相當(dāng)于從頭開(kāi)始查詢所有的數(shù)據(jù)庫(kù)記錄
      clean_run => "false"

      # 是否將 字段(column) 名稱轉(zhuǎn)小寫
      lowercase_column_names => "false"

      columns_charset => {
        "message"=> "UTF-8"
        "name"=> "UTF-8"
      }
    }
}

# 此處我不做過(guò)濾處理,如果需要，也可參考elk安裝那篇
filter {}

output {
    # 輸出到elasticsearch的配置
    # 注意這里對(duì)type判斷，若加載多個(gè)配置文件，要有這個(gè)判斷才不會(huì)互相影響
    if[type] == "jdbc" {
      elasticsearch {
          hosts => ["192.168.31.45:9200"]
          index => "test"

          # 將"_id"的值設(shè)為mysql的autoid字段
          # 注意這里的id，如果多個(gè)表輸出到同一個(gè)index，它們的id有重復(fù)的，則這里的 document_id 要修改成不重復(fù)的，否則會(huì)覆蓋數(shù)據(jù)
          document_id => "%{id}"
          template_overwrite => true
      }
    }
    # 這里輸出調(diào)試，正式運(yùn)行時(shí)可以注釋掉
    stdout {
        codec => json_lines
    }
}

啟動(dòng)：

./bin/logstash -f ./myconf/jdbc.conf

問(wèn)題：編碼錯(cuò)誤

[2019-05-15T21:38:10,051][ERROR][logstash.outputs.elasticsearch] An unknown error occurred sending a bulk request to Elasticsearch. We will retry indefinitely {:error_message=>""xE8" from ASCII-8BIT to UTF-8", :error_class=>"LogStash::Json::GeneratorError", :backtrace=>["/home/mt/Desktop/logstash-7.0.1/logstash-core/lib/logstash/json.rb:27:in `jruby_dump"", "/home/mt/Desktop/logstash-7.0.1/vendor/bundle/jruby/2.5.0/gems/logstash-output-elasticsearch-10.0.2-java/lib/logstash/outputs/elasticsearch/http_client.rb:119:in `block in bulk"", "org/jruby/RubyArray.java:2577:in `map"", "/home/mt/Desktop/logstash-7.0.1/vendor/bundle/jruby/2.5.0/gems/logstash-output-elasticsearch-10.0.2-java/lib/logstash/outputs/elasticsearch/http_client.rb:119:in `block in bulk"", "org/jruby/RubyArray.java:1792:in `each"", "/home/mt/Desktop/logstash-7.0.1/vendor/bundle/jruby/2.5.0/gems/logstash-output-elasticsearch-10.0.2-java/lib/logstash/outputs/elasticsearch/http_client.rb:117:in `bulk"", "/home/mt/Desktop/logstash-7.0.1/vendor/bundle/jruby/2.5.0/gems/logstash-output-elasticsearch-10.0.2-java/lib/logstash/outputs/elasticsearch/common.rb:286:in `safe_bulk"", "/home/mt/Desktop/logstash-7.0.1/vendor/bundle/jruby/2.5.0/gems/logstash-output-elasticsearch-10.0.2-java/lib/logstash/outputs/elasticsearch/common.rb:191:in `submit"", "/home/mt/Desktop/logstash-7.0.1/vendor/bundle/jruby/2.5.0/gems/logstash-output-elasticsearch-10.0.2-java/lib/logstash/outputs/elasticsearch/common.rb:159:in `retrying_submit"", "/home/mt/Desktop/logstash-7.0.1/vendor/bundle/jruby/2.5.0/gems/logstash-output-elasticsearch-10.0.2-java/lib/logstash/outputs/elasticsearch/common.rb:38:in `multi_receive"", "org/logstash/config/ir/compiler/OutputStrategyExt.java:118:in `multi_receive"", "org/logstash/config/ir/compiler/AbstractOutputDelegatorExt.java:101:in `multi_receive"", "/home/mt/Desktop/logstash-7.0.1/logstash-core/lib/logstash/java_pipeline.rb:235:in `block in start_workers""]}

解決：

對(duì)各個(gè)字段設(shè)置字符集：

      columns_charset => {
        "message"=> "UTF-8"
        "name"=> "UTF-8"
        "payload"=> "UTF-8"
      }

問(wèn)題：加載多個(gè)配置文件進(jìn)行數(shù)據(jù)遷移，有重復(fù)id

運(yùn)行多個(gè)實(shí)例

創(chuàng)建一個(gè)配置文件的文件夾，使用-f命令加載此文件即可。

./bin/logstash -f ./myconf/

想要維持一個(gè)長(zhǎng)期后臺(tái)運(yùn)行的 logstash，你需要同時(shí)在命令前面加 nohup，后面加 &。

logstash mysql 準(zhǔn)實(shí)時(shí)同步到 elasticsearch

logstash-input-jdbc同步mysql數(shù)據(jù)到elasticsearch

MySQL到Elasticsearch的同步之路

discuss.elastic.co/t/filter-er…

www.elastic.co/guide/en/lo…

discuss.elastic.co/t/character…

Logstash Multiple Pipelines 使用

Logstash 最佳實(shí)踐

logstash-input-jdbc同時(shí)同步多個(gè)表