QQ 咨询
喝杯饮料
微信交流
wx公众号
432

Logstash同步MySQL一对多关联表到Elasticsearch父子文档
如何文章对你有用的话,就点击左边悬浮按钮,请小编喝杯饮料吧
时间: 2021-12-20 20:09:20  作者:北桥苏  阅读:(56)


前言:

         目前大部分业务开发中,ElasticSearch主要还是用来做搜索。而支撑搜索功能的数据结构比较单一,不会有数据嵌套或者多种关联之类的。尽管没有,但是有些小众需求可能还会有一对多查询的场景。为了实现和MySQL的Join类似的查询方式,以下以ES的父子文档方式储存,并详细演示Logstash如何将MySQL的多张有关联的表同步到ES的父子文档。

 

手动演示:


         以下以restful方式创建父子文档索引,并以简单的方式查询类似join的数据返回。下面所有演示的索引名称都为 "my_join_index"。

1. 创建父子关联索引

PUT my_join_index
{
  "mappings": {
    "properties": {
        "my_join_field": { 
          "type": "join",
          "relations": {
            "question": "answer" 
          }
        }
      }
  }
}

2. 创建父文档

PUT my_join_index/_doc/1?refresh
{
  "text": "This is a question",
  "my_join_field": "question" 
}


PUT my_join_index/_doc/2?refresh
{
  "text": "This is another question",
  "my_join_field": "question"
}

3. 创建子文档

PUT my_join_index/_doc/3?routing=1&refresh 
{
  "text": "This is an answer",
  "my_join_field": {
    "name": "answer", 
    "parent": "1" 
  }
}


PUT my_join_index/_doc/4?routing=1&refresh
{
  "text": "This is another answer2",
  "my_join_field": {
    "name": "answer",
    "parent": "2"
  }
}

4. 全局检索

GET my_join_index/_search
{
  "query": {
    "match_all": {}
  },
  "sort": ["_id"]
}

5. 根据父文档查找子文档

GET my_join_index/_search
{
    "query": {
        "has_parent" : {
            "parent_type" : "question",
            "query" : {
                "match" : {
                    "text" : "This is"
                }
            }
        }
    }
}

6. 根据子文档查找父文档

GET my_join_index/_search
{
"query": {
        "has_child" : {
            "type" : "answer",
            "query" : {
                "match" : {
                    "text" : "This is question"
                }
            }
        }
    }
}

7. Join聚合

GET my_join_index/_search
{
  "query": {
    "parent_id": { 
      "type": "answer",
      "id": "1"
    }
  },
  "aggs": {
    "parents": {
      "terms": {
        "field": "my_join_field#question", 
        "size": 10
      }
    }
  },
  "script_fields": {
    "parent": {
      "script": {
         "source": "doc['my_join_field#question']" 
      }
    }
  }
}

8. 单条联合查询, 可以是一条父文档对应多个子文档

GET my_join_index/_search
{
  "query": {
    "bool": {
      "must": [
        {
          "match": {
              "title": "历史圈"
          }
        },
        {
          "has_child": {
            "type": "answer",
            "query": {
              "match": {
                "text":"是的"
              }
            },
            "inner_hits":{}
          }
        }
      ]
    }
  }
}

    

Logstash同步:


              以下以文章分类表和文章表为例,二者系一对多的关系。同步文档时,文章分类作为父文档,文章作为子文档,关联字段为 “my_join_field”。

 

1.  创建有父子文档的索引

PUT hhyp_article
{
  "mappings": {
    "properties": {
      "my_join_field": { 
        "type": "join",
        "relations": {
          "article_cate": "article" 
        }
      }
    }
  }
}

2. 配置同步代码



input {

    stdin {
		
    }
	
	jdbc {
	
      # mysql 数据库链接,shop为数据库名
      jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/rebuild?characterEncoding=UTF-8&useSSL=false"
	  
      # 用户名和密码
      jdbc_user => "root"
	  
      jdbc_password => "root"
	  
      # 驱动
      jdbc_driver_library => "E:/2setsoft/1dev/logstash-7.8.0/mysqletc/mysql-connector-java-5.1.7-bin.jar"
	  
      # 驱动类名
      jdbc_driver_class => "com.mysql.jdbc.Driver"
	  
      jdbc_paging_enabled => "true"
      jdbc_page_size => "50000"
	  
	  parameters => {"number" => "200"}
	  
	  statement => "SELECT * FROM `hhyp_article` WHERE delete_time = 0"
	  
	  # 是否将字段名转换为小写,默认true(如果有数据序列化、反序列化需求,建议改为false);
      lowercase_column_names => false
		
	  # Value can be any of: fatal,error,warn,info,debug,默认info;
	  sql_log_level => warn
		
	  
      # 设置监听间隔  各字段含义(由左至右)分、时、天、月、年,全部为*默认含义为每分钟都更新
      schedule => "* * * * *"
	  
      # 索引类型
      type => "article"
    }
	
	jdbc {
	
      # mysql 数据库链接,shop为数据库名
      jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/rebuild?characterEncoding=UTF-8&useSSL=false"
	  
      # 用户名和密码
      jdbc_user => "root"
      jdbc_password => "root"
	  
      # 驱动
      jdbc_driver_library => "E:/2setsoft/1dev/logstash-7.8.0/mysqletc/mysql-connector-java-5.1.7-bin.jar"
	  
      # 驱动类名
      jdbc_driver_class => "com.mysql.jdbc.Driver"
	  
      jdbc_paging_enabled => "true"
      jdbc_page_size => "50000"
	  
	  parameters => {"number" => "200"}
	  
	  statement => "SELECT * FROM `hhyp_article_cate` WHERE delete_time = 0"
	  
	  # 是否将字段名转换为小写,默认true(如果有数据序列化、反序列化需求,建议改为false);
      lowercase_column_names => false
		
	  # Value can be any of: fatal,error,warn,info,debug,默认info;
	  sql_log_level => warn
	  
      # 设置监听间隔  各字段含义(由左至右)分、时、天、月、年,全部为*默认含义为每分钟都更新
      schedule => "* * * * *"
	  
      # 索引类型
      type => "article_cate"
    }	
}
 
filter {
  if [type]=="article_cate" {
	  mutate {
			add_field => { "my_join_field" => "article_cate" }
	  }
	}
	
	if [type]=="article" {
	  mutate {
			add_field => {"[my_join_field][name]" => "article"}
			
			#catalog_id 子表的父id
			add_field => {"[my_join_field][parent]" => "%{cid}"}  
	  }
	}
	
}
 
output {
	
	if[type] == "article_cate" {
		elasticsearch {
			hosts => "localhost:9200"
			index => "hhyp_article"
			document_type => "_doc"
			document_id => "%{id}"
		}
    }
	  
    if[type] == "article" {
		elasticsearch {
			hosts => "localhost:9200"
			index => "hhyp_article"
			document_type => "_doc"
			document_id => "%{id}"
			routing => "%{cid}"
		}
	}
	
    stdout {
        codec => json_lines
    }
	
}

3. 运行命令开始同步

bin\logstash -f mysql\mysql.conf

4.  通过搜索父文档标题查询子文档数据






用户登录


     QQ登录

我的标签


随机文章

  • 浅析网站被挂马的代码其实我网站中马应该可以追朔到去年的八月份,当…… 查看详情
  • 自制微擎AI面相识别算术阈值有时在朋友圈或其他地方会看到一些AI面相的分…… 查看详情
  • 关于ThinkPHP5.1 redis驱动的cache的两个小笔记在ThinkPHP5.1中cache类,可以…… 查看详情

友情链接





Copyright© 2016-2021 北桥苏 & 版权所有    赣ICP备16002525号-1
如有侵权、不妥之处,请联系站长的客服并出示版权证明以便删除。敬请谅解! 站长客服联系方式:QQ2652364582