一、前置條件

? ? ? ? 1.? linux，已經(jīng)搭建好的logstash+es+kibana【系列版本7.0X】，es 的plugs中安裝ik分詞器

ES版本：

?Logstash版本：

?（以上部署，都是運(yùn)維同事搞的，我不會(huì)部署，同事給力）

二、編寫(xiě)Logstash.sh 執(zhí)行文件

1、在Logstash安裝目錄下【/usr/share/logstash】，新建XX.sh，內(nèi)容如下：

/usr/share/logstash/bin/logstash --path.data /usr/share/logstash/case-conf -e 'input {
 jdbc {
    jdbc_driver_library => "/var/local/logstash/etc/lib/mysql-connector-java-8.0.15.jar"
    jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://IP:端口號(hào)/數(shù)據(jù)庫(kù)?serverTimezone=Asia/Shanghai&autoReconnect=true&allowMultiQueries=true&zeroDateTimeBehavior=convertToNull&characterEncoding=utf8&useSSL=false&tinyInt1isBit=false"
    jdbc_user => "用戶名"
    jdbc_password => "密碼"
    schedule => "* * * * *"
    use_column_value => true
    tracking_column => "update_at"
    tracking_column_type => "numeric"
    last_run_metadata_path => "/usr/share/logstash/case-last"
statement_filepath => "/usr/share/logstash/case_.sql"
  }
}

output {
  elasticsearch {
    hosts => ["es1:9206","es2:9207","es3:9208"]
	action=>"index"
    index => "case"
    document_id => "%{case_id}"
    template => "/usr/share/logstash/template-case.json"
	template_name=>"template-case.json"
	template_overwrite=>true
    }
}'

2. 在Logstash安裝目錄下【/usr/share/logstash】,新建case.sql文件：查詢字段，根據(jù)業(yè)務(wù)要求書(shū)寫(xiě)，不需要全字段查詢

SELECT * FROM 表名稱(chēng) where update_at > :sql_last_value

where條件中的，update_at 是表更新時(shí)間，與case.sh 中的?tracking_column 強(qiáng)關(guān)聯(lián)；默認(rèn)logstash 每分鐘執(zhí)行一次case.sql ,執(zhí)行后會(huì)生成一個(gè)case-last文件，里面記錄最后一次執(zhí)行時(shí)間；

3.? 關(guān)于output，配置項(xiàng)，使用了自定義創(chuàng)建索引的模板方式，在表中有數(shù)據(jù)為前提，執(zhí)行case.sh 會(huì)用到模板創(chuàng)建索引；如果不配置，es會(huì)默認(rèn)給創(chuàng)建mapping，默認(rèn)的話，分詞都是英文分詞器； 執(zhí)行效果如下：sh? case.sh后：

?注意細(xì)節(jié)：?

output {
? elasticsearch {
? ? hosts => []
?? ?action=>"index"
? ? index => "case"? ?//索引名稱(chēng)
? ? document_id => "%{case_id}"? //動(dòng)態(tài)數(shù)據(jù)，數(shù)據(jù)ID
? ? template => "/usr/share/logstash/template-case.json"? ?//模板文件路徑
?? ?template_name=>"template-case.json"? ? ? ? ? ? //模板名稱(chēng)
?? ?template_overwrite=>true? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?
? ? }
}

4.?在Logstash安裝目錄下【/usr/share/logstash】，新建template-case.json，內(nèi)容如下：

{
  "template" : "case*",
  "settings" : {
    "index.refresh_interval" : "5s",
    "number_of_replicas":"1",
    "number_of_shards":"1"
  },
  "mappings": {
    "date_detection": false,
    "numeric_detection": false,
    "dynamic_templates": [
      {
        "integers": {
          "match_mapping_type": "long",
          "mapping": {
            "type": "integer"
          }
        }
      },
      {
        "strings": {
          "match_mapping_type": "string",
          "unmatch": "*_en",
          "mapping": {
            "type": "text",
            "analyzer":"ik_max_word",
            "search_analyzer": "ik_smart",
            "fields": {
              "raw": {
                "type":  "keyword",
                "ignore_above": 100
              }
            }
          }
        }
      }
    ]
  }
}

注意：

1、模板名稱(chēng)和case.sh 中的index 匹配；模板中的名稱(chēng)后必須帶個(gè)*：? "template" : "case*",

2、模板比較簡(jiǎn)單，關(guān)閉了日期動(dòng)態(tài)檢測(cè)和數(shù)字格式動(dòng)態(tài)檢測(cè)；不然創(chuàng)建索引的時(shí)候格式會(huì)亂；尤其是日期；比如數(shù)據(jù)庫(kù)中字段是varchar，但是存的是yyyy-MM-dd hh:mm:ss 日期格式的話，es創(chuàng)建索引是date格式的；

? ? "date_detection": false,
? ? "numeric_detection": false,

3、string格式的字段，默認(rèn)給轉(zhuǎn)出text, 并且使用ik分詞器；排除了_en結(jié)尾的字段，存英文的字段建表的時(shí)候注意下，使用默認(rèn)分詞就好；

? ?"unmatch": "*_en"

三、Kibana驗(yàn)證

1、GET case_/_mapping：

{
  "case" : {
    "mappings" : {
      "dynamic_templates" : [
        {
          "integers" : {
            "match_mapping_type" : "long",
            "mapping" : {
              "type" : "integer"
            }
          }
        },
        {
          "strings" : {
            "unmatch" : "*_en",
            "match_mapping_type" : "string",
            "mapping" : {
              "analyzer" : "ik_max_word",
              "fields" : {
                "raw" : {
                  "ignore_above" : 100,
                  "type" : "keyword"
                }
              },
              "search_analyzer" : "ik_smart",
              "type" : "text"
            }
          }
        }
      ],
      "date_detection" : false,
      "numeric_detection" : false,
      "properties" : {
        "@timestamp" : {
          "type" : "text",
          "fields" : {
            "raw" : {
              "type" : "keyword",
              "ignore_above" : 100
            }
          },
          "analyzer" : "ik_max_word",
          "search_analyzer" : "ik_smart"
        },
        "@version" : {
          "type" : "text",
          "fields" : {
            "raw" : {
              "type" : "keyword",
              "ignore_above" : 100
            }
          },
          "analyzer" : "ik_max_word",
          "search_analyzer" : "ik_smart"
        },
        "apply_education" : {
          "type" : "text",
          "fields" : {
            "raw" : {
              "type" : "keyword",
              "ignore_above" : 100
            }
          },
          "analyzer" : "ik_max_word",
          "search_analyzer" : "ik_smart"
        },  
        "apply_major_en" : {
          "type" : "text",
          "fields" : {
            "keyword" : {
              "type" : "keyword",
              "ignore_above" : 256
            }
          }
        },
        "apply_school_name" : {
          "type" : "text",
          "fields" : {
            "raw" : {
              "type" : "keyword",
              "ignore_above" : 100
            }
          },
          "analyzer" : "ik_max_word",
          "search_analyzer" : "ik_smart"
        },
        "apply_school_name_en" : {
          "type" : "text",
          "fields" : {
            "keyword" : {
              "type" : "keyword",
              "ignore_above" : 256
            }
          }
        },
     
        "begin_learn_time_at" : {
          "type" : "integer"
        },
  
        "case_id" : {
          "type" : "integer"
        },     
   
        "case_result_time_at" : {
          "type" : "integer"
        },
    
        "country_name" : {
          "type" : "text",
          "fields" : {
            "raw" : {
              "type" : "keyword",
              "ignore_above" : 100
            }
          },
          "analyzer" : "ik_max_word",
          "search_analyzer" : "ik_smart"
        },
  
   
    
        "school_id" : {
          "type" : "integer"
        },
          
        "update_at" : {
          "type" : "integer"
        }
      }
    }
  }
}

查看，映射字段是否滿足要求；

2.? 驗(yàn)證索引分詞結(jié)果：

GET case/_analyze
{
? "field": "apply_education",
? "text": "馬斯特里赫特大學(xué)"?
}

分詞結(jié)果：

{
  "tokens" : [
    {
      "token" : "馬斯特里赫特",
      "start_offset" : 0,
      "end_offset" : 6,
      "type" : "CN_WORD",
      "position" : 0
    },
    {
      "token" : "馬斯",
      "start_offset" : 0,
      "end_offset" : 2,
      "type" : "CN_WORD",
      "position" : 1
    },
    {
      "token" : "特里",
      "start_offset" : 2,
      "end_offset" : 4,
      "type" : "CN_WORD",
      "position" : 2
    },
    {
      "token" : "赫",
      "start_offset" : 4,
      "end_offset" : 5,
      "type" : "CN_CHAR",
      "position" : 3
    },
    {
      "token" : "特大",
      "start_offset" : 5,
      "end_offset" : 7,
      "type" : "CN_WORD",
      "position" : 4
    },
    {
      "token" : "大學(xué)",
      "start_offset" : 6,
      "end_offset" : 8,
      "type" : "CN_WORD",
      "position" : 5
    }
  ]
}

驗(yàn)證完成；

四、相關(guān)資料

1.?Elastic：開(kāi)發(fā)者上手指南_elastic.show_Elastic 中國(guó)社區(qū)官方博客的博客-CSDN博客

2.?Mutate filter plugin | Logstash Reference [8.8] | Elastic

五、注意事項(xiàng)

1. 此種方法，只能針對(duì)表數(shù)值為邏輯刪除的情況，若業(yè)務(wù)是物理刪除，則需要同步刪除索引中的數(shù)據(jù)；

2. 所有update 操作，都需要同時(shí)修改 update_at 字段文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-516610.html

到了這里，關(guān)于ELK增量同步數(shù)據(jù)【MySql-＞ES】的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！