logstash 增量同步mysql的一对多记录到ES的Join类型

小灰灰 2022-04-11 04:37 260阅读 0赞

> ES 我用的是6.5.1

#### mysql 数据模型 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2ODYzOTk_size_16_color_FFFFFF_t_70]

*  esdatabase 的建库语句

CREATE DATABASE IF NOT EXISTS esdatabase DEFAULT CHARSET utf8 COLLATE utf8_general_ci;

*  brand建表语句

CREATE TABLE `brand` (
      `ability_id`  int auto_increment NOT NULL,
      `brand_name` varchar(128) DEFAULT NULL,
      PRIMARY KEY (`ability_id`)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

*  product 建表语句

CREATE TABLE `product` (
      `product_id`  int auto_increment NOT NULL,
      `brand_name` varchar(128) DEFAULT NULL,
      `product_name` varchar(128) DEFAULT NULL,
      `product_price`   float(5,2)  DEFAULT NULL,
      `product_color` varchar(128) DEFAULT NULL,
      PRIMARY KEY (`product_id`)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

> product 表没有创建外键。这里还是看个人观点吧，添加外键有可能影响性能，但是需要应用维护一致性。总之，要维护product和brand的一致性。

*  插入测试数据

insert brand(brand_name) value ('Nike'),('Address'),('360');
    insert product(brand_name,product_name,product_price,product_color) value ('1','shoe',31,'red'),('1','cloth',32,'green'),('1','trouser',55,'black');
    insert product(brand_name,product_name,product_price,product_color) value ('3','shoe',11,'white'),('3','cloth',32,'green'),('3','trouser',18,'pink');
    insert product(brand_name,product_name,product_price,product_color) value ('4','shoe',111,'red'),('4','cloth',132,'purple'),('4','trouser',155,'black');

#### ES 创建索引 ####

http://x.x.x.x:9200/testjon

{
      "mappings": {
        "doc": {
          "dynamic": "strict",
          "properties": {
            "type": {
              "type": "text"
            },
             "ability_id": {
              "type": "integer"
            },
             "product_id": {
              "type": "integer"
            },
                    "brand_name": {
              "type": "integer"
            },
            "brand_name": {
              "type": "text"
            },
            "product_name": {
              "type": "text"
            },
            "product_price": {
              "type": "float"
            },
            "product_color": {
              "type": "text"
            },
            "product_list": {
              "type": "join",
              "relations": {
                "brand": "product"
              }
            }
          }
        }
      }
    }

> type ability\_id product\_id brand\_name 字段其实没有什么用途，我是被迫添加的。  
> 因为output阶段需要使用这些字段，所以不能再filter阶段删除掉，但是output阶段又没有删除字段的功能  
> 这个问题困惑着我，还请各位大神指点一下。。。。

#### logstash 配置文件 ####

> logstash 用了jdbc和alter两个插件，要是大家不方便安装的话，QQ群里有我安装好的tar包，QQ群号在底部。

input {
        jdbc {
          type => "brand"
          jdbc_connection_string => "jdbc:mysql://1.1.1.1:3306/esdatabase" 
          jdbc_user => "esuser"    
          jdbc_password => "123.com"
          jdbc_driver_library => "/app/logstash/mysql-connector-java-5.1.47.jar"
          jdbc_driver_class => "com.mysql.jdbc.Driver"
          jdbc_paging_enabled => "true"
          jdbc_page_size => "1000"
          last_run_metadata_path => "/app/logstash/logstash_jdbc_last_run"
          use_column_value => true
          tracking_column => "ability_id"
          statement => "select * from brand where ability_id> :sql_last_value"
          schedule => "*/1 * * * *"
        }
     
        jdbc {
          type => "product"
          jdbc_connection_string => "jdbc:mysql://1.1.1.1::3306/esdatabase" 
          jdbc_user => "esuser"    
          jdbc_password => "123.com"
          jdbc_driver_library =>  "/app/logstash/mysql-connector-java-5.1.47.jar"
          jdbc_driver_class => "com.mysql.jdbc.Driver"
          last_run_metadata_path => "/app/logstash/logstash_jdbc_last_run_02"
          use_column_value => true
          tracking_column => "product_id"
          jdbc_paging_enabled => "true"
          jdbc_page_size => "1000"
          statement => " select * from product where product_id >  :sql_last_value"
          schedule => "*/2 * * * *"
        }
    }
    filter {
            if[type] == "brand"{
                      alter {
                        add_field  => {
                              "[product_list][name]"=> "brand"
                        }
                      }
            }
            if[type] == "product"{
                      alter {
                        add_field  => {
                             "[product_list][name]" => "product"
                             "[product_list][parent]" =>  "%{brand_name}"
                        }
                      }
            }
             mutate {
                remove_field => [ "@timestamp","@version"]
              }
    }
    output{ 
            if [type] == "brand" { 
                    elasticsearch { 
                            hosts => "1.1.1.1::9200" 
                            index => "testjoin" 
                            document_id => "brand_%{ability_id}"
                    } 
            }
            if [type]== "product"  { 
                            elasticsearch { 
                                    hosts => "1.1.1.1::9200" 
                                    index => "testjoin" 
                                    document_id => "product_%{product_id}"
                                    routing => "brand_%{brand_name}"
                                      } 
                                    } 
         }

##### 说几点 #####

*  schedule，建议brand表也就是较小的表应该在较大的表之前全部同步到ES，防止建立Join类型时找不到父文档，所以这里的时间请大家自己衡量。
 *  last\_run\_metadata\_path use\_column\_value tracking\_column sql\_last\_value这几个结合起来用于增量同步，具体字段的功能可以去官网看看
 *  document\_id 我这里把父文档设置为 brand\_ 开头，把子文档设置为 product\_ 开头，这样可以避免 id重复的问题，因为在mysql中这两张表的id都是从1开始的
 *  routing 这个字段必须有的，保证子文档和父文档在一个分片上

#### 查看结果 ####

启动logstash

logstash -f  testJoin.yml

##### 观察ES结果 #####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2ODYzOTk_size_16_color_FFFFFF_t_70 1]

##### 插入几条数据，看看是不是增量 #####

insert brand(brand_name) value ('lining');
    insert product(brand_name,product_name,product_price,product_color) value ('1','shoe',678,'blue');
     insert product(brand_name,product_name,product_price,product_color) value ('4','shoe',278,'black');

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2ODYzOTk_size_16_color_FFFFFF_t_70 2]

--------------------

> 先写到这里了，有问题进QQ群630300475

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2ODYzOTk_size_16_color_FFFFFF_t_70]: /images/20220411/80173a3f4a554c7a8e0848cdb599a926.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2ODYzOTk_size_16_color_FFFFFF_t_70 1]: /images/20220411/11819a2c4ea84261889397874081a26b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2ODYzOTk_size_16_color_FFFFFF_t_70 2]: /images/20220411/4617385dbd674e87855166f6b30a4d45.png