Logstash同步ES

Java
499
0
0
2022-04-20
Netty粘包&半包解决方案
上一篇
拉仇恨！webhook + 企业微信给同事做了个代码提交监听工具
下一篇

Logstash 简介

Logstash 是一个数据流引擎：

它是用于数据物流的开源流式 ETL（Extract-Transform-Load）引擎

在几分钟内建立数据流管道

具有水平可扩展及韧性且具有自适应缓冲

不可知的数据源

具有200多个集成和处理器的插件生态系统

使用 Elastic Stack 监视和管理部署

Logstash 是如何工作的？

Laravel

Logstash 包含3个主要部分： 输入（inputs），过滤器（filters）和输出（outputs）。 你必须定义这些过程的配置才能使用 Logstash，尽管不是每一个都必须的。在有些情况下，我们可以甚至没有过滤器。在过滤器的部分，它可以对数据源的数据进行分析，丰富，处理等等。

安装

docker pull logstash:7.12.1
----------------------------------
vim logstash.conf

配置文件如下

input {
 stdin { }
    jdbc {
        #注意mysql连接地址一定要用ip，不能使用localhost等
        jdbc_connection_string => "jdbc:mysql://172.18.12.1:3306/lmrs_2008_shops"
        jdbc_user => "root"
        jdbc_password => "root" 
        #数据库重连尝试
        connection_retry_attempts => "3" 
        #数据库连接可用校验超时时间，默认为3600s
        jdbc_validation_timeout => "3600" 
        #这个jar包的地址是容器内的地址
        jdbc_driver_library => "/etc/logstash/pipeline/mysql-connector-java-8.0.24.jar"
        jdbc_driver_class => "com.mysql.jdbc.Driver" 
        #开启分页查询（默认是false）
        jdbc_paging_enabled => "true" 
        #单次分页查询条数（默认100000，字段较多的话，可以适当调整这个数值）
        jdbc_page_size => "50000" 
        #执行的sql语句
        statement => "select a.id,a.`name`,a.long_name,a.brand_id,a.three_category_id as category_id,a.shop_id,a.price,a.sold_count,a.review_count,a.`status`,a.create_time,a.last_time,b.`name` as category,b.path from lmrs_products as a LEFT JOIN lmrs_product_categorys as b on a.three_category_id = b.id where a.id > :sql_last_value" 
        #需要记录查询结果某字段的值时，此字段为true，否则默认tracking_colum为timestamp的值
        use_column_value => true 
        #是否将字段名转为小写，默认为true（如果具备序列化或者反序列化，建议设置为false）
        lowercase_column_names => false 
        #需要记录的字段，同于增量同步，需要是数据库字段
        tracking_column => id 
        #记录字段的数据类型
        tracking_column_type => numeric
        #上次数据存放位置
        record_last_run => true 
        #上一个sql_last_value的存放路径，必须在文件中指定字段的初始值
        last_run_metadata_path => "/etc/logstash/pipeline/products.txt" 
        #是否清除last_run_metadata_path的记录，需要增量同步这个字段的值必须为false
        clean_run => false 
        #同步的频率（分 时 天 月 年）默认为每分钟同步一次
        schedule => "* * * * *" 
        type => "_doc"
    }
    jdbc {
        #注意mysql连接地址一定要用ip，不能使用localhost等
        jdbc_connection_string => "jdbc:mysql://172.18.12.1:3306/lmrs_2008_shops"
        jdbc_user => "root"
        jdbc_password => "root" 
        #数据库重连尝试
        connection_retry_attempts => "3" 
        #数据库连接可用校验超时时间，默认为3600s
        jdbc_validation_timeout => "3600" 
        #这个jar包的地址是容器内的地址
        jdbc_driver_library => "/etc/logstash/pipeline/mysql-connector-java-8.0.24.jar"
        jdbc_driver_class => "com.mysql.jdbc.Driver" 
        #开启分页查询（默认是false）
        jdbc_paging_enabled => "true" 
        #单次分页查询条数（默认100000，字段较多的话，可以适当调整这个数值）
        jdbc_page_size => "50000" 
        #执行的sql语句
        statement => "select c.*,d.`name`as category,d.path as category_path from (select b.id,a.`name`,b.`name` as `value`,a.sort as attribute_sort,b.sort as attribute_value_sort,a.category_id from lmrs_attributes as a LEFT JOIN lmrs_attribute_values as b on a.id = b.attribute_id) as c LEFT JOIN lmrs_product_categorys as d on c.category_id = d.id where c.id
> :sql_last_value" 
        #需要记录查询结果某字段的值时，此字段为true，否则默认tracking_colum为timestamp的值
        use_column_value => true 
        #是否将字段名转为小写，默认为true（如果具备序列化或者反序列化，建议设置为false）
        lowercase_column_names => false 
        #需要记录的字段，同于增量同步，需要是数据库字段
        tracking_column => id 
        #记录字段的数据类型
        tracking_column_type => numeric
        #上次数据存放位置
        record_last_run => true 
        #上一个sql_last_value的存放路径，必须在文件中指定字段的初始值
        last_run_metadata_path => "/etc/logstash/pipeline/attributes.txt" 
        #是否清除last_run_metadata_path的记录，需要增量同步这个字段的值必须为false
        clean_run => false 
        #同步的频率（分 时 天 月 年）默认为每分钟同步一次
        schedule => "* * * * *" 
        type => "attributes"
    }
 }
 
filter {
    jdbc_streaming {
         jdbc_driver_library => "/etc/logstash/pipeline/mysql-connector-java-8.0.24.jar"
         jdbc_driver_class => "com.mysql.jdbc.Driver"
         jdbc_connection_string => "jdbc:mysql://172.18.12.1:3306/lmrs_2008_shops"
         jdbc_user => "root"
         jdbc_password => "root"
         parameters => { "sensor_identifier" => "id"}
         statement => "select `name`,price from lmrs_product_skus where product_id= :sensor_identifier"
         target => "skus"
    }
 
    jdbc_streaming {
        jdbc_driver_library => "/etc/logstash/pipeline/mysql-connector-java-8.0.24.jar"
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        jdbc_connection_string => "jdbc:mysql://172.18.12.1:3306/lmrs_2008_shops"
        jdbc_user => "root"
        jdbc_password => "root"
        parameters => { "sensor_identifiers" => "id"}
        statement => "select c.`name`,f.`name` as `value` from (select b.`name`,b.id FROM lmrs_product_attribute_values as a LEFT JOIN lmrs_attributes as b on a.attribute_id= b.id where a.product_id = :sensor_identifiers) as c LEFT JOIN(select d.attribute_id,d.`name` from lmrs_attribute_values as d LEFT JOIN lmrs_product_attribute_values as e on d.id = e.attribute_value_id where product_id = :sensor_identifiers) as f on c.id = f.attribute_id GROUP BY f.`name`"
        target => "attributes"
    }
}
 
output {
    if [type] == "_doc" {
        elasticsearch {
            #注意mysql连接地址一定要用ip，不能使用localhost等
            hosts => "172.18.12.6:9200"
            index => "products"
            document_type => "_doc"
            document_id => "%{id}"
        }
    }
    if [type] == "attributes" {
        elasticsearch {
            #注意mysql连接地址一定要用ip，不能使用localhost等
            hosts => "172.18.12.6:9200"
            index => "attributes"
            document_type => "_doc"
            document_id => "%{id}"
        }
    }
    stdout {
        codec => json_lines
    }
}

索引

PUT /products/
{
  "mappings": {
    "properties": {
      "name":{
        "type": "text",
        "analyzer": "ik_smart"
      },
      "long_name":{
        "type": "text",
        "analyzer": "ik_smart"
      },
      "brand_id":{
        "type": "integer"
      },
      "category_id":{
        "type":"integer"
      },
      "category":{
        "type": "keyword"
      },
      "category_path":{
        "type": "keyword"
      },
      "shop_id":{
        "type":"integer"
      },
      "price":{
        "type":"scaled_float",
        "scaling_factor":100
      },
      "sold_count":{
        "type":"integer"
      },
      "review_count":{
        "type":"integer"
      },
      "status":{
        "type":"integer"
      },
      "create_time" : {
          "type" : "date"
      },
      "last_time" : {
          "type" : "date"
      },
      "skus":{
        "type": "nested",
        "properties": {
          "name":{
            "type":"text",
            "analyzer": "ik_smart"
          },
          "price":{
            "type":"scaled_float",
            "scaling_factor":100
          }
        }
      },
      "attributes":{
          "type": "nested",
          "properties": {
            "name": { "type": "keyword" },
            "value": { "type": "keyword"}
          }
      }
    }
  }
}
 
 
 
PUT /attributes
{
  "mappings": {
    "properties": {
      "name": {
        "type": "keyword"
      },
      "value": {
        "type": "keyword"
      },
      "category_id": {
        "type": "integer"
      },
      "attribute_sort": {
        "type": "integer"
      },
      "attribute_value_sort": {
        "type": "integer"
      },
      "category": {
        "type": "keyword"
      },
      "category_path": {
        "type": "text"
      }
    }
  }
}

es查询结果

{
  "took" : 4,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 2,
      "relation" : "eq"
    },
    "max_score" : 0.18232156,
    "hits" : [
      {
        "_index" : "products",
        "_type" : "_doc",
        "_id" : "2",
        "_score" : 0.18232156,
        "_source" : {
          "status" : 1,
          "category_id" : 440,
          "long_name" : "HUAWEI Mate Book 14 32GB 1TB 触屏 集显",
          "path" : "-425-438-",
          "id" : 2,
          "type" : "_doc",
          "shop_id" : 2,
          "review_count" : 222,
          "attributes" : [
            {
              "value" : "翡冷翠",
              "name" : "颜色"
            },
            {
              "value" : "冰霜银",
              "name" : "颜色"
            },
            {
              "value" : "星际蓝",
              "name" : "颜色"
            },
            {
              "value" : "R5/32GB/1TB 触屏",
              "name" : "配置"
            },
            {
              "value" : "R7/32GB/1TB 触屏",
              "name" : "配置"
            },
            {
              "value" : "集成显卡",
              "name" : "显卡"
            },
            {
              "value" : "官方标配",
              "name" : "类型"
            }
          ],
          "brand_id" : 2,
          "skus" : [
            {
              "name" : "翡冷翠 R5/32GB/1TB 触屏 触屏 集成显卡 官方标配",
              "price" : 7999
            },
            {
              "name" : "翡冷翠 R7/32GB/1TB 触屏 触屏 集成显卡 官方标配",
              "price" : 9999
            },
            {
              "name" : "冰霜银 R5/32GB/1TB 触屏 触屏 集成显卡 官方标配",
              "price" : 7999
            },
            {
              "name" : "冰霜银 R7/32GB/1TB 触屏 触屏 集成显卡 官方标配",
              "price" : 9999
            },
            {
              "name" : "星际蓝 R5/32GB/1TB 触屏 触屏 集成显卡 官方标配",
              "price" : 7999
            },
            {
              "name" : "星际蓝 R7/32GB/1TB 触屏 触屏 集成显卡 官方标配",
              "price" : 9999
            }
          ],
          "@timestamp" : "2021-06-03T08:55:01.124Z",
          "category" : "笔记本电脑",
          "name" : "HUAWEI Mate Book 14",
          "@version" : "1",
          "create_time" : "2021-05-28T20:02:02.000Z",
          "sold_count" : 222,
          "last_time" : "2021-05-28T21:10:04.000Z",
          "price" : 7999.0
        }
      },
      {
        "_index" : "products",
        "_type" : "_doc",
        "_id" : "1",
        "_score" : 0.18232156,
        "_source" : {
          "status" : 1,
          "category_id" : 440,
          "long_name" : "HUAWEI Mate Book 13 16GB 512GB 触屏 集显",
          "path" : "-425-438-",
          "id" : 1,
          "type" : "_doc",
          "shop_id" : 1,
          "review_count" : 1111,
          "attributes" : [
            {
              "value" : "皓月银",
              "name" : "颜色"
            },
            {
              "value" : "深空灰",
              "name" : "颜色"
            },
            {
              "value" : "樱粉金",
              "name" : "颜色"
            },
            {
              "value" : "I5/16GB/512GB 触屏",
              "name" : "配置"
            },
            {
              "value" : "I7/16GB/512GB 触屏",
              "name" : "配置"
            },
            {
              "value" : "集成显卡",
              "name" : "显卡"
            },
            {
              "value" : "官方标配",
              "name" : "类型"
            }
          ],
          "brand_id" : 1,
          "skus" : [
            {
              "name" : "皓月银 I5/16GB/512GB 触屏 集成显卡 官方标配",
              "price" : 6299
            },
            {
              "name" : "皓月银 I7/16GB/512GB 触屏 集成显卡 官方标配",
              "price" : 6599
            },
            {
              "name" : "深空灰 I5/16GB/512GB 触屏 集成显卡 官方标配",
              "price" : 6299
            },
            {
              "name" : "深空灰 I7/16GB/512GB 触屏 集成显卡 官方标配",
              "price" : 6599
            },
            {
              "name" : "樱粉金 I5/16GB/512GB 触屏 集成显卡 官方标配",
              "price" : 6299
            },
            {
              "name" : "樱粉金 I7/16GB/512GB 触屏 集成显卡 官方标配",
              "price" : 6599
            }
          ],
          "@timestamp" : "2021-06-03T08:55:01.122Z",
          "category" : "笔记本电脑",
          "name" : "HUAWEI Mate Book 13",
          "@version" : "1",
          "create_time" : "2021-05-25T15:12:09.000Z",
          "sold_count" : 111,
          "last_time" : "2021-05-25T15:12:14.000Z",
          "price" : 6299.0
        }
      }
    ]
  }
}
Netty粘包&半包解决方案
上一篇
拉仇恨！webhook + 企业微信给同事做了个代码提交监听工具
下一篇