1. 代码100分首页
  2. 代码基础

全量同步和增量同步_es多个索引关联查询

小小码 • • 代码基础 • 阅读 149
一键激活最新全家桶

全量同步和增量同步_es多个索引关联查询MySQLmom同步MySQL到es

全量同步多张db表到ES同一个索引[数据库教程]

一、演示场景:

演示的场景主要是解决MySQL多张业务大表进行多表join查询效率低下的问题。
通过把MySQL的多张大表的数据同步到同一个ES索引中。(也就是有多表字段合并到es一张宽表来解决MySQL多表join效率低下的问题)

1.1、演示环境

自建MySQL服务5.7.22

ES单实例版本6.2.4

服务器python环境2.7.5

部署同步服务mysqlmom程序

具体安装部署此处忽略。有需要可以查看本博客找

二、MySQLmom具体配置文件

[[email protected] ~]# cat /data1/soft/mysqlsmom01/test_mom/init_config.py
# coding=utf-8
STREAM = "INIT"
# 修改数据库连接
CONNECTION = {
    ‘host‘: ‘172.16.0.197‘,
    ‘port‘: 3306,
    ‘user‘: ‘click_rep‘,
    ‘passwd‘: ‘jwtest123456‘
}
# 一次同步 BULK_SIZE 条数据到elasticsearch，不设置该配置项默认为1
BULK_SIZE = 50000
# 修改elasticsearch节点
#NODES = [{"host": "127.0.0.1", "port": 9200}]
NODES = [{"host": "172.16.0.247", "port": 9999}]
TASKS = [
# 同步stdb01.test03到es:
    {
        "stream": {
            "database": "test_db",  # 在此数据库执行sql语句
            "sql": "select * from test01",  # 将该sql语句选中的数据同步到 elasticsearch
            # "pk": {"field": "id", "type": "char"}  # 当主键id的类型是字符串时
        },
        "jobs": [
            {
                "actions": ["insert", "update"],
                "pipeline": [
                    {"only_fields": {"fields": ["id", "username"]}}, # 只同步 id 和 username字段
                    {"set_id": {"field": "id"}}  # 默认设置 id字段的值 为elasticsearch中的文档id
                ],
                "dest": {
                    "es": {
                        "action": "upsert",
                        "index": "test01_company_index",   # 设置 index
                        "type": "test01",          # 设置 type
                        "nodes": NODES
                    }
                }
            }
        ]
    },
    {
        "stream": {
            "database": "test_db",  # 在此数据库执行sql语句
            "sql": "select * from company_staff",  # 将该sql语句选中的数据同步到 elasticsearch
            # "pk": {"field": "id", "type": "char"}  # 当主键id的类型是字符串时
        },
        "jobs": [
            {
                "actions": ["insert", "update"],
                "pipeline": [
                    {"only_fields": {"fields": ["id", "company_name", "company_staff", "channel", "url"]}}, # 只同步 id 和 username字段
                    {"set_id": {"field": "id"}}  # 默认设置 id字段的值 为elasticsearch中的文档id
                ],
                "dest": {
                    "es": {
                        "action": "upsert",
                        "index": "test01_company_index",   # 设置 index
                        "type": "test01",          # 设置 type
                        "nodes": NODES
                    }
                }
            }
        ]
    }

]
# CUSTOM_ROW_HANDLERS = "./my_handlers.py"
# CUSTOM_ROW_FILTERS = "./my_filters.py"

?

三、mysql测试表建表语句,表数据以及同步程序启动

mysql测试表建表语句和表数据如下：

[email protected] 16:16:  [test_db]> show create table company_staffG
*************************** 1. row ***************************
       Table: company_staff
Create Table: CREATE TABLE `company_staff` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT ‘id‘,
  `company_name` varchar(255) COLLATE utf8mb4_unicode_ci NOT NULL DEFAULT ‘‘ COMMENT ‘公司名‘,
  `company_staff` varchar(255) COLLATE utf8mb4_unicode_ci NOT NULL DEFAULT ‘‘ COMMENT ‘人员规模‘,
  `channel` varchar(10) COLLATE utf8mb4_unicode_ci NOT NULL DEFAULT ‘‘ COMMENT ‘来源‘,
  `url` varchar(255) COLLATE utf8mb4_unicode_ci NOT NULL DEFAULT ‘‘ COMMENT ‘url‘,
  `create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT ‘创建时间‘,
  `update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT ‘更新时间‘,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=8 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT=‘namelist人员规模表‘
1 row in set (0.00 sec)

[email protected] 16:17:  [test_db]> select * from company_staff;
+----+-----------------------------------------------------------------------------+---------------+-------------+-----------------------------------------+---------------------+---------------------+
| id | company_name                                                                | company_staff | channel     | url                                     | create_time         | update_time         |
+----+-----------------------------------------------------------------------------+---------------+-------------+-----------------------------------------+---------------------+---------------------+
|  1 | 永兴东润（中国）服饰有限公司北京海淀第四儿童服饰店                          | liepin        | 100-499人   | https://www.liepin.com/company/8321725/ | 2021-06-19 17:21:57 | 2021-06-19 17:21:57 |
|  2 | 东(中国）服饰有限公司北京海淀第四儿童服饰店                                 | liepin        | 100-499人   | https://www.liepin.com/company/8321725/ | 2021-06-19 17:21:57 | 2021-06-19 17:21:57 |
|  3 | 永兴东润(中国）服饰有限公司北京海淀第四儿童服饰店                           | liepin        | 100-499人   | https://www.liepin.com/company/8321725/ | 2021-06-19 17:21:57 | 2021-06-19 17:21:57 |
|  4 | 润(中国）                                                                   | liepin        | 100-499人   | https://www.liepin.com/company/8321725/ | 2021-06-19 17:21:57 | 2021-06-19 17:21:57 |
+----+-----------------------------------------------------------------------------+---------------+-------------+-----------------------------------------+---------------------+---------------------+
4 rows in set (0.00 sec)

[email protected] 16:17:  [test_db]> show create table test01G
*************************** 1. row ***************************
       Table: test01
Create Table: CREATE TABLE `test01` (
  `id` int(8) NOT NULL AUTO_INCREMENT,
  `username` varchar(20) COLLATE utf8_unicode_ci NOT NULL,
  `password` varchar(20) COLLATE utf8_unicode_ci NOT NULL,
  `create_time` varchar(20) COLLATE utf8_unicode_ci NOT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=5 DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci
1 row in set (0.00 sec)


[email protected] 16:17:  [test_db]> select * from test01;
+----+----------+------------+---------------------+
| id | username | password   | create_time         |
+----+----------+------------+---------------------+
|  1 | tomcat   | xiaohuahua | 2021-07-03 23:51:17 |
|  2 | php      | xiao       | 2021-07-03 23:53:36 |
|  3 | fix      | xiao       | 2021-07-03 23:53:49 |
|  4 | java     | bai        | 2021-07-03 23:54:01 |
+----+----------+------------+---------------------+
4 rows in set (0.00 sec)



[[email protected] mysqlsmom01]# mom run -c ./test_mom/binlog_config.py >mysqlmom.log 2>&1 &
2021-08-08 16:23:57,873 root         INFO     {"username": "tomcat", "_id": 1, "id": 1}
2021-08-08 16:23:57,874 root         INFO     {"username": "php", "_id": 2, "id": 2}
2021-08-08 16:23:57,874 root         INFO     {"username": "fix", "_id": 3, "id": 3}
2021-08-08 16:23:57,874 root         INFO     {"username": "java", "_id": 4, "id": 4}
2021-08-08 16:23:57,975 elasticsearch INFO     POST http://172.16.0.247:9999/_bulk [status:200 request:0.101s]
2021-08-08 16:23:57,979 root         INFO     {"url": "https://www.liepin.com/company/8321725/", "company_staff": "liepin", "company_name": "u6c38u5174u4e1cu6da6uff08u4e2du56fduff09u670du9970u6709u9650u516cu53f8u5317u4eacu6d77u6dc0u7b2cu56dbu513fu7ae5u670du9970u5e97", "_id": 1, "id": 1, "channel": "100-499u4eba "}
2021-08-08 16:23:57,979 root         INFO     {"url": "https://www.liepin.com/company/8321725/", "company_staff": "liepin", "company_name": "u4e1c(u4e2du56fduff09u670du9970u6709u9650u516cu53f8u5317u4eacu6d77u6dc0u7b2cu56dbu513fu7ae5u670du9970u5e97", "_id": 2, "id": 2, "channel": "100-499u4eba "}
2021-08-08 16:23:57,979 root         INFO     {"url": "https://www.liepin.com/company/8321725/", "company_staff": "liepin", "company_name": "u6c38u5174u4e1cu6da6(u4e2du56fduff09u670du9970u6709u9650u516cu53f8u5317u4eacu6d77u6dc0u7b2cu56dbu513fu7ae5u670du9970u5e97", "_id": 3, "id": 3, "channel": "100-499u4eba "}
2021-08-08 16:23:57,979 root         INFO     {"url": "https://www.liepin.com/company/8321725/", "company_staff": "liepin", "company_name": "u6da6(u4e2du56fduff09", "_id": 4, "id": 4, "channel": "100-499u4eba "}
2021-08-08 16:23:58,007 elasticsearch INFO     POST http://172.16.0.247:9999/_bulk [status:200 request:0.027s]

real	0m0.637s
user	0m0.447s
sys	0m0.061s

全量同步启动命令如下:

?

[[email protected] mysqlsmom01]# mom run -c ./test_mom/init_config.py >mysqlmom.log 2>&1 &
2021-08-08 16:23:57,873 root         INFO     {"username": "tomcat", "_id": 1, "id": 1}
2021-08-08 16:23:57,874 root         INFO     {"username": "php", "_id": 2, "id": 2}
2021-08-08 16:23:57,874 root         INFO     {"username": "fix", "_id": 3, "id": 3}
2021-08-08 16:23:57,874 root         INFO     {"username": "java", "_id": 4, "id": 4}
2021-08-08 16:23:57,975 elasticsearch INFO     POST http://172.16.0.247:9999/_bulk [status:200 request:0.101s]
2021-08-08 16:23:57,979 root         INFO     {"url": "https://www.liepin.com/company/8321725/", "company_staff": "liepin", "company_name": "u6c38u5174u4e1cu6da6uff08u4e2du56fduff09u670du9970u6709u9650u516cu53f8u5317u4eacu6d77u6dc0u7b2cu56dbu513fu7ae5u670du9970u5e97", "_id": 1, "id": 1, "channel": "100-499u4eba "}
2021-08-08 16:23:57,979 root         INFO     {"url": "https://www.liepin.com/company/8321725/", "company_staff": "liepin", "company_name": "u4e1c(u4e2du56fduff09u670du9970u6709u9650u516cu53f8u5317u4eacu6d77u6dc0u7b2cu56dbu513fu7ae5u670du9970u5e97", "_id": 2, "id": 2, "channel": "100-499u4eba "}
2021-08-08 16:23:57,979 root         INFO     {"url": "https://www.liepin.com/company/8321725/", "company_staff": "liepin", "company_name": "u6c38u5174u4e1cu6da6(u4e2du56fduff09u670du9970u6709u9650u516cu53f8u5317u4eacu6d77u6dc0u7b2cu56dbu513fu7ae5u670du9970u5e97", "_id": 3, "id": 3, "channel": "100-499u4eba "}
2021-08-08 16:23:57,979 root         INFO     {"url": "https://www.liepin.com/company/8321725/", "company_staff": "liepin", "company_name": "u6da6(u4e2du56fduff09", "_id": 4, "id": 4, "channel": "100-499u4eba "}
2021-08-08 16:23:58,007 elasticsearch INFO     POST http://172.16.0.247:9999/_bulk [status:200 request:0.027s]

real	0m0.637s
user	0m0.447s
sys	0m0.061s

?

图示如下:
watermark,size_14,text_QDUxQ1RP5Y2a5a6i,color_FFFFFF,t_100,g_se,x_10,y_10,shadow_20,type_ZmFuZ3poZW5naGVpdGk=

到此时，全量同步多张db表数据到ES同一个索引演示完成

版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。
转载请注明出处: https://daima100.com/5970.html

赞 (0)
小小码小小码
0 0
上一篇 2023-04-20
下一篇 2023-04-20

相关推荐

  • 一致性就是相合性_数据库中事物的四大特性 代码基础

    一致性就是相合性_数据库中事物的四大特性

    一致性就是相合性_数据库中事物的四大特性事务的特性 dddddddd 原子性，事务中所有操作是不可再分割的原子单位。事务中所有操作要么全部执行成功，要么全部执行失败; 一致性，事务执行后，数据库状态与其它业务规则保持一致; 隔离性，指在并…

    小小码 小小码
    2023-02-05
    253
  • 思维进阶 查询连续签到几天的用户「终于解决」 代码基础

    思维进阶 查询连续签到几天的用户「终于解决」

    思维进阶 查询连续签到几天的用户「终于解决」select user_name,sign_date,IF(@pre=user_name,@rownum:=@rownum+1,@rownum:=1), @pre:=user_name from (…

    小小码 小小码
    2023-03-09
    150
  • Hash存储模型、B-Tree存储模型、LSM存储模型介绍[通俗易懂] 代码基础

    Hash存储模型、B-Tree存储模型、LSM存储模型介绍[通俗易懂]

    Hash存储模型、B-Tree存储模型、LSM存储模型介绍[通俗易懂]每一种数据存储系统，对应有一种存储模型，或者叫存储引擎。我们今天要介绍的是三种比较流行的存储模型，分别是：Hash存储模型B-Tree存储模型LSM存储模型不同存储模型的应用情况1、Hash存储模型r

    小小码 小小码
    2023-01-26
    156
  • Timeout expired. The timeout period elapsed prior to obtaining a connection from the pool 代码基础

    Timeout expired. The timeout period elapsed prior to obtaining a connection from the pool

    Timeout expired. The timeout period elapsed prior to obtaining a connection from the pool如果应用程序遇到了下面错误信息，那么意味着连接池（connection pool）的连接数量由于一些原因导致其超过了Max Pool Size参数的限制。 英文错误信息： Timeout expire

    小小码 小小码
    2023-02-06
    143
  • 浅谈达梦数据库的兼容模式 代码基础

    浅谈达梦数据库的兼容模式

    浅谈达梦数据库的兼容模式达梦数据库号称国产数据库第一品牌，在某些方面还是有它的优势的，作为IT攻城狮，我从事软件行业已经有10多年了，oracle、mysq、sqlserver都已经用的很熟了，这2年做了几个国产化相关的项…

    小小码 小小码
    2023-02-03
    180
  • Python中的title方法 代码基础

    Python中的title方法

    Python中的title方法在Python中，字符串是一种常见的数据类型，而其中一个很有用的字符串方法是title()方法。通俗地说，title()方法可以将字符串中的每个单词的首字母都变成大写，而其他字母则变成小写。

    admin admin
    2024-04-21
    70
  • 使用jQuery实现DOM元素的查找和操作 代码基础

    使用jQuery实现DOM元素的查找和操作

    使用jQuery实现DOM元素的查找和操作在Web前端开发中，操作DOM元素是非常重要的一部分。jQuery是一个非常流行的JavaScript库，提供了丰富的API，方便我们进行DOM操作。本文将介绍如何使用jQuery实现DOM元素的查找和操作。

    admin admin
    2024-05-19
    64
  • Python计算方差 代码基础

    Python计算方差

    Python计算方差方差是概率论和统计学中最基础的概念之一，它可以测量一组数据样本的离散程度。在数据分析中，方差是一个非常重要的指标，它允许我们定量地了解数据的变化情况。Python作为一种流行的数据分析语言，拥有很多处理统计学问题的工具和库，下面我们就来介绍一下Python如何计算方差。

    admin admin
    2024-05-05
    59

发表回复

您的电子邮箱地址不会被公开。 必填项已用*标注