Logstash 是 Elastic 技術(shù)棧中的一個技術(shù)，它是一個數(shù)據(jù)采集引擎，可以從數(shù)據(jù)庫采集數(shù)據(jù)到 ES 中?？梢酝ㄟ^設(shè)置 自增 ID 主鍵 或 更新時間 來控制數(shù)據(jù)的自動同步：

• 自增 ID 主鍵：Logstatsh 會有定時任務(wù)，如果發(fā)現(xiàn)有主鍵的值大于先前同步記錄的主鍵值，就會將對應(yīng)的增量數(shù)據(jù)同步到 ES 中
• 更新時間：其實原理與主鍵類似，不過如果設(shè)置使用主鍵作為依據(jù)的話，那么數(shù)據(jù)庫的數(shù)據(jù)更新就不會被識別從而更新到 ES 中。

一、安裝

1. 下載地址：https://www.elastic.co/cn/downloads/past-releases#logstash

• 注：使用Logstatsh的版本號與elasticsearch版本號需要保持一致

2. 上傳并解壓

先上傳到服務(wù)器 /home/software/ 下，然后解壓，并將解壓后的文件夾移動到 /usr/local/ 下

tar -zxvf logstash-7.6.2.tar.gz

mv logstash-7.6.2 /usr/local/

二、配置

1. 首先在 Elasticsearch 中創(chuàng)建一個索引：didiok-items

2. 在 /usr/local/logstash-7.6.2/ 下創(chuàng)建文件夾 sync/

將數(shù)據(jù)庫驅(qū)動 mysql-connector-java-5.1.41.jar 包上傳到??/usr/local/logstash-7.6.2/sync/ 下，

cd /usr/local/logstash-7.6.2/
mkdir sync
cd sync/

3. 編寫數(shù)據(jù)同步的SQL腳本

 SELECT
     i.id as id,
     i.item_name as itemName,
     i.sell_counts as sellCounts,
     ii.url as imgUrl,
     tempSpec.price_discount as price,
     i.updated_time as updated_time
 FROM
     items i
 LEFT JOIN
     items_img ii
 on
     i.id = ii.item_id
 LEFT JOIN
     (SELECT item_id,MIN(price_discount) as price_discount from items_spec GROUP BY item_id) tempSpec
 on
     i.id = tempSpec.item_id
 WHERE
     ii.is_main = 1
     and
     i.updated_time >= :sql_last_value

# :sql_last_value 是 logstash 每次同步完成之后保存的的邊界值，這里保存的是 updated_time ，用于下次數(shù)據(jù)同步時，大于等于 updated_time 的數(shù)據(jù)才會進行同步

將sql腳本?保存到 /usr/local/logstash-7.6.2/sync/didiok-items.sql 文件中?

4.? 在?sync/ 下創(chuàng)建 配置文件?logstash-db-sync.conf，內(nèi)容如下：

input {
    jdbc {
        # 設(shè)置 MySql/MariaDB 數(shù)據(jù)庫url以及數(shù)據(jù)庫名稱
        jdbc_connection_string => "jdbc:mysql://192.168.1.6:3306/didiok-shop-dev?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true"
        # 用戶名和密碼
        jdbc_user => "root"
        jdbc_password => "root"
        # 數(shù)據(jù)庫驅(qū)動所在位置，可以是絕對路徑或者相對路徑
        jdbc_driver_library => "/usr/local/logstash-7.6.2/sync/mysql-connector-java-5.1.41.jar"
        # 驅(qū)動類名
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        # 開啟分頁
        jdbc_paging_enabled => "true"
        # 分頁每頁數(shù)量，可以自定義
        jdbc_page_size => "10000"
        # 執(zhí)行的sql文件路徑
        statement_filepath => "/usr/local/logstash-7.6.2/sync/didiok-items.sql"
        # 設(shè)置定時任務(wù)間隔  含義：分、時、天、月、年，全部為*默認含義為每分鐘跑一次任務(wù)
        schedule => "* * * * *"
        # 索引類型
        type => "_doc"
        # 是否開啟記錄上次追蹤的結(jié)果，也就是上次更新的時間，這個會記錄到 last_run_metadata_path 的文件
        use_column_value => true
        # 記錄上一次追蹤的結(jié)果值
        last_run_metadata_path => "/usr/local/logstash-7.6.2/sync/track_time"
        # 如果 use_column_value 為true， 配置本參數(shù)，追蹤的 column 名，可以是自增id或者時間
        tracking_column => "updated_time"
        # tracking_column 對應(yīng)字段的類型
        tracking_column_type => "timestamp"
        # 是否清除 last_run_metadata_path 的記錄，true則每次都從頭開始查詢所有的數(shù)據(jù)庫記錄
        clean_run => false
        # 數(shù)據(jù)庫字段名稱大寫轉(zhuǎn)小寫
        lowercase_column_names => false
    }
}
output {
    elasticsearch {
        # es地址
        hosts => ["192.168.1.187:9200"]
        # 同步的索引名
        index => "didiok-items"
        # 設(shè)置_docID和數(shù)據(jù)庫中的id相同
        document_id => "%{id}"

    }
    # 日志輸出
    stdout {
        codec => json_lines
    }}

5. 啟動 logstash

cd /usr/local/logstash-7.6.2/bin

./logstash -f /usr/local/logstash-7.6.2/sync/logstash-db-sync.conf

如果啟動過程中報錯如下：

報這個錯是虛擬機內(nèi)存不足，是因為這個Logstash要的內(nèi)存太大了。這個插件在初始化的時候要了一個g的內(nèi)存，但是虛擬機沒有這么多的內(nèi)存，所以把這個內(nèi)存改小就行了。

（?。。〉?，這樣好像會導(dǎo)致 logstash 啟動不了，還是想辦法擴大內(nèi)存吧 ，下面的方法僅供參考）

修改?/usr/local/logstash-7.6.2/config/jvm.options ,這里原來是 1g，修改成 256m：

三、在自定義模板中配置中文分詞器

首先在 ES 中創(chuàng)建索引 didiok-items，之后啟動 logstash，然后再進行以下操作。

1.?查看Logstash默認模板

請求方式：GET

路徑：http://localhost:9200/_template/logstash

2. 將查詢出來的模板復(fù)制出來，進行修改如下（這里只修改了3處）：

{
? ? "order": 0,
? ? "version": 1,            # 修改1
? ? "index_patterns": ["*"], # 修改2
? ? "settings": {
? ? ? ? "index": {
? ? ? ? ? ? "refresh_interval": "5s"
? ? ? ? }
? ? },
? ? "mappings": {
? ? ? ? "_default_": {
? ? ? ? ? ? "dynamic_templates": [
? ? ? ? ? ? ? ? {
? ? ? ? ? ? ? ? ? ? "message_field": {
? ? ? ? ? ? ? ? ? ? ? ? "path_match": "message",
? ? ? ? ? ? ? ? ? ? ? ? "match_mapping_type": "string",
? ? ? ? ? ? ? ? ? ? ? ? "mapping": {
? ? ? ? ? ? ? ? ? ? ? ? ? ? "type": "text",
? ? ? ? ? ? ? ? ? ? ? ? ? ? "norms": false
? ? ? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? },
? ? ? ? ? ? ? ? {
? ? ? ? ? ? ? ? ? ? "string_fields": {
? ? ? ? ? ? ? ? ? ? ? ? "match": "*",
? ? ? ? ? ? ? ? ? ? ? ? "match_mapping_type": "string",
? ? ? ? ? ? ? ? ? ? ? ? "mapping": {
? ? ? ? ? ? ? ? ? ? ? ? ? ? "type": "text",
? ? ? ? ? ? ? ? ? ? ? ? ? ? "norms": false,
? ? ? ? ? ? ? ? ? ? ? ? ? ? "analyzer": "ik_max_word",  # 修改3 加入中文分詞器
? ? ? ? ? ? ? ? ? ? ? ? ? ? "fields": {
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? "keyword": {
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? "type": "keyword",
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? "ignore_above": 256
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? }
? ? ? ? ? ? ],
? ? ? ? ? ? "properties": {
? ? ? ? ? ? ? ? "@timestamp": {
? ? ? ? ? ? ? ? ? ? "type": "date"
? ? ? ? ? ? ? ? },
? ? ? ? ? ? ? ? "@version": {
? ? ? ? ? ? ? ? ? ? "type": "keyword"
? ? ? ? ? ? ? ? },
? ? ? ? ? ? ? ? "geoip": {
? ? ? ? ? ? ? ? ? ? "dynamic": true,
? ? ? ? ? ? ? ? ? ? "properties": {
? ? ? ? ? ? ? ? ? ? ? ? "ip": {
? ? ? ? ? ? ? ? ? ? ? ? ? ? "type": "ip"
? ? ? ? ? ? ? ? ? ? ? ? },
? ? ? ? ? ? ? ? ? ? ? ? "location": {
? ? ? ? ? ? ? ? ? ? ? ? ? ? "type": "geo_point"
? ? ? ? ? ? ? ? ? ? ? ? },
? ? ? ? ? ? ? ? ? ? ? ? "latitude": {
? ? ? ? ? ? ? ? ? ? ? ? ? ? "type": "half_float"
? ? ? ? ? ? ? ? ? ? ? ? },
? ? ? ? ? ? ? ? ? ? ? ? "longitude": {
? ? ? ? ? ? ? ? ? ? ? ? ? ? "type": "half_float"
? ? ? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? }
? ? ? ? ? ? }
? ? ? ? }
? ? },
? ? "aliases": {}
}

之后將其保存為? /usr/local/logstash-7.6.2/sync/logstash-ik.json

3. 在?/usr/local/logstash-7.6.2/sync/logstash-db-sync.conf 文件中進行修改，加入以下內(nèi)容：

# 定義模板名稱
template_name => "myik"
# 模板所在位置
template => "/usr/local/logstash-7.6.2/sync/logstash-ik.json"
# 重寫模板
template_overwrite => true
# 默認為true，false關(guān)閉logstash自動管理模板功能，如果自定義模板，則設(shè)置為false
manage_template => false

4.?重新運行Logstash進行同步

./logstash -f /usr/local/logstash-7.6.2/sync/logstash-db-sync.conf

中文分詞器沒有設(shè)置成功？?

試試下面的解決方案：

先通過postman請求 http://192.168.1.187:9200/_template/logstash，獲取的json放入logstash-ik.json中，然后在 /usr/local/logstash-7.6.2/sync/logstash-db-sync.conf 文件中設(shè)置manage_template => true，然后啟動logstash，啟動后 ES 的 didiok-items?索引是不正確的。

刪除索引，并重新創(chuàng)建索引?didiok-items。

然后 postman 調(diào) http://192.168.1.187:9200/_template/myik 拿到的 myik 的 json 重新放入logstash-ik.json中，配置 logstash-db-sync.conf? 改為 manage_template => false，

再次啟動logstash就能在?didiok-items?的mapping中顯示中文分詞器了。

文章來源地址http://www.zghlxwxcb.cn/news/detail-456340.html

到了這里，關(guān)于Logstash數(shù)據(jù)同步的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！