ELK增量同步数据【MySql-＞ES】-Toy模板网

这篇具有很好参考价值的文章主要介绍了ELK增量同步数据【MySql-＞ES】。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

一、前置条件

1. linux，已经搭建好的logstash+es+kibana【系列版本7.0X】，es 的plugs中安装ik分词器

ES版本： ELK增量同步数据【MySql-＞ES】,ELK,elk,elasticsearch

Logstash版本： ELK增量同步数据【MySql-＞ES】,ELK,elk,elasticsearch

（以上部署，都是运维同事搞的，我不会部署，同事给力）

二、编写Logstash.sh 执行文件

1、在Logstash安装目录下【/usr/share/logstash】，新建XX.sh，内容如下：

/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:端口号/数据库?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文件：查询字段，根据业务要求书写，不需要全字段查询

SELECT * FROM 表名称 where update_at > :sql_last_value

where条件中的，update_at 是表更新时间，与case.sh 中的 tracking_column 强关联；默认logstash 每分钟执行一次case.sql ,执行后会生成一个case-last文件，里面记录最后一次执行时间； ELK增量同步数据【MySql-＞ES】,ELK,elk,elasticsearch

3. 关于output，配置项，使用了自定义创建索引的模板方式，在表中有数据为前提，执行case.sh 会用到模板创建索引；如果不配置，es会默认给创建mapping，默认的话，分词都是英文分词器；执行效果如下：sh case.sh后：

ELK增量同步数据【MySql-＞ES】,ELK,elk,elasticsearch

注意细节：

output {
elasticsearch {
hosts => []
   action=>"index"
index => "case" //索引名称
document_id => "%{case_id}" //动态数据，数据ID
template => "/usr/share/logstash/template-case.json" //模板文件路径
   template_name=>"template-case.json" //模板名称
   template_overwrite=>true
}
}

4. 在Logstash安装目录下【/usr/share/logstash】，新建template-case.json，内容如下：

{
  "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、模板名称和case.sh 中的index 匹配；模板中的名称后必须带个*： "template" : "case*",

2、模板比较简单，关闭了日期动态检测和数字格式动态检测；不然创建索引的时候格式会乱；尤其是日期；比如数据库中字段是varchar，但是存的是yyyy-MM-dd hh:mm:ss 日期格式的话，es创建索引是date格式的；

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

3、string格式的字段，默认给转出text, 并且使用ik分词器；排除了_en结尾的字段，存英文的字段建表的时候注意下，使用默认分词就好；

"unmatch": "*_en"

三、Kibana验证

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. 验证索引分词结果：

GET case/_analyze
{
"field": "apply_education",
"text": "马斯特里赫特大学"
}

分词结果：

{
  "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" : "大学",
      "start_offset" : 6,
      "end_offset" : 8,
      "type" : "CN_WORD",
      "position" : 5
    }
  ]
}

验证完成；