wansho-GitBook
GitHubAbout
Search…
README
读书
现代操作系统
流畅的 Python
计算机网络-自顶向下
数据结构与算法
其他书籍
计算机笔记
算法
database
zen-of-db
MySQL-101
MySQL-102
MySQL-103
Elasticsearch
Logstash
Oracle-101
Mysql-Leetcode
其他数据库产品
书籍-设计数据密集型应用
Questions
密码学与网络安全
计算机网络相关知识
C++ 开发
Python 学习
Java 学习笔记
JavaWeb 学习笔记
前端开发
NLP
linux
github
代码与总结
开发工具入门与技术栈
技术讲座
生活和思考
blog
关于健康
求知若渴虚心若愚
安心立命
Powered By GitBook
Elasticsearch
[TOC]

技术栈

MySQL 有的功能，elastic search 一般都有。(elastic search 没有 acid 特性)
  • 增量数据迁移 mysql => es: alibaba canal
    canal 有服务端(slaver管道)、adapter(消费者) 和 admin 三个主要程序
    配置特别麻烦，不适合多现场部署，版本兼容性也不咋地，太重了
  • 全量数据迁移：logstash

视频教程

elasticsearch笔记链接：https://pan.baidu.com/s/1dvrgjdiwf09cOhBiL7ezWQ 提取码：8888 Elasticsearch大数据搜索引擎链接：https://pan.baidu.com/s/14Tc8c6ZXZLZ61i-rKLXxGQ 提取码：8888 第二版：Elasticsearch 24 道链接：https://pan.baidu.com/s/15DadWyGblRA1Tenn7JthEg 提取码：8888

数据高可用

image-20210801094139464
高可用，就是数据冗余，elastic search 会进行数据冗余，副本分片是只读的，主要用来做搜索。
集群健康值
green-yellow-red
查看当前集群不健康的原因：get _cluster/allocation/explain
灰色的分片，是因为我们只有一个节点，数据是不可能有副本的，这违背了高可用原则。
怎么把黄色告警删除？把这个索引的副本分片设置为 0:
1
put index_name/_settings
2
{
3
"number_of_replicas": 1
4
}
Copied!
image-20210801095253754

index, shards, replicas

​
​
elastic search 不能以 root 用户启动。

基于canal的MySQL=>ES数据同步方案

  1. 1.

    概念

    canal [kə'næl]，译意为水道/管道/沟渠，主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费
    早期阿里巴巴因为杭州和美国双机房部署，存在跨机房同步的业务需求，实现方式主要是基于业务 trigger 获取增量变更。从 2010 年开始，业务逐步尝试数据库日志解析获取增量变更进行同步，由此衍生出了大量的数据库增量订阅和消费业务。
  2. 2.

    应用场景

    Elasticsearch 不支持事务。 ES通常在分布式系统架构中承担“搜索引擎”的角色，一般来说解决词类问题，可以把ES和支持ACID特性的关系型数据库结合起来使用。首先把对数据的更（增删改）操作在RDB中执行，然后把这些动作同步到Elasticsearch。 通过这种方式，你将受益于数据库 ACID 事务支持，并且在 Elasticsearch 中以正确的顺序产生变更。 并发在关系数据库中得到了处理。
    以MySQL为例，如果要把数据从同步至ES，canal + binary log就是常用的一种增量解决方案。
  3. 3.

    原理

    ​
    ​
    MySQL主备复制原理
    • MySQL master 将数据变更写入二进制日志( binary log, 其中记录叫做二进制日志事件binary log events，可以通过 show binlog events 进行查看)
    • MySQL slave 将 master 的 binary log events 拷贝到它的中继日志(relay log)
    • MySQL slave 重放 relay log 中事件，将数据变更反映它自己的数据
    canal 工作原理
    • canal 模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave ，向 MySQL master 发送dump 协议
    • MySQL master 收到 dump 请求，开始推送 binary log 给 slave (即 canal )
    • canal 解析 binary log 对象(原始为 byte 流)
      ​
      ​
  4. 4.

    优势

    • 准实时性
    • 性能好
    • 一劳永逸
  5. 5.

    玩法

    1. 1.
      环境：
      ​ Java和ES兼容性：https://www.elastic.co/cn/support/matrix#matrix_jvm
      • JDK：1.8
      • Elasticsearch：7.x
      • MySQL： 5.7
      • Canal： 1.1.4
    2. 2.
      下载：
      Github： https://github.com/alibaba/canal/
      Github 咻咻咻： https://github.com/fhefh2015/Fast-GitHub
    3. 3.
      步骤：
      1. 1.
        保证Elasticsearch服务可用
      2. 2.
        保证MySQL服务可用
      3. 3.
        开启MySQL的binary log（主备模式）
        • 配置：
        1
        server_id = 1 #开启主从模式后每个MySQL节点的id
        2
        log-bin = mysql-bin #bin-log的存储位置
        3
        binlog-format = ROW #选择存储binlog日志方式为ROW模式
        Copied!
        • 重启MySQL服务
        • 验证是否开启成功
          1
          SHOW VARIABLES LIKE 'log_bin';
          2
          log_bin ON #开启
          Copied!
      4. 4.
        canal-deployer
        • 配置conf/example/instance.properties
          1
          #canal示例的slaveId
          2
          canal.instance.mysql.slaveId=1234
          3
          #mysql地址
          4
          canal.instance.master.address= 127.0.0.1:3306
          5
          #用户名
          6
          canal.instance.dbUsername = root
          7
          #密码
          8
          canal.instance.dbPassword = 123456
          9
          #指定需要同步的数据库
          10
          canal.instance.defaultDatabaseName = msb_order
          11
          #指定编码方式
          12
          canal.instance.connectionCharset = UTF-8
          13
          #监控的是所有数据库，所有的表改动都会监控到，这样可能会浪费不少性能，可能我只想监控的是某一个数据库下的表。
          14
          # .*\\..*表示监控所有数据库，canal\\..*表示监控canal数据库
          15
          canal.instance.filter.regex = .\*\\\\..\*
          Copied!
        • 启动： ./startup.sh（Linux）
        • 验证：demo
      5. 5.
        canal-admin
        • 配置conf/application.yml
          1
          server:
          2
          port: 8089
          3
          spring:
          4
          jackson:
          5
          date-format: yyyy-MM-dd HH:mm:ss
          6
          time-zone: GMT+8
          7
          ​
          8
          spring.datasource:
          9
          address: 127.0.0.1:3306
          10
          database: canal_manager
          11
          username: root
          12
          password: 123456
          13
          driver-class-name: com.mysql.jdbc.Driver
          14
          url: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=false
          15
          hikari:
          16
          maximum-pool-size: 30
          17
          minimum-idle: 1
          18
          canal:
          19
          adminUser: admin
          20
          adminPasswd: admin
          Copied!
        • 启动管理服务
        • 访问服务：server_ip:8089
      6. 6.
        canal-adapter
        • 配置conf/application.yml
        1
        server:
        2
        port: 8081
        3
        spring:
        4
        jackson:
        5
        date-format: yyyy-MM-dd HH:mm:ss
        6
        time-zone: GMT+8
        7
        default-property-inclusion: non_null
        8
        ​
        9
        canal.conf:
        10
        mode: tcp #tcp kafka rocketMQ rabbitMQ
        11
        flatMessage: true
        12
        zookeeperHosts:
        13
        syncBatchSize: 1000
        14
        retries: 0
        15
        timeout:
        16
        accessKey:
        17
        secretKey:
        18
        consumerProperties:
        19
        # canal tcp consumer
        20
        canal.tcp.server.host: 127.0.0.1:11111
        21
        canal.tcp.zookeeper.hosts:
        22
        canal.tcp.batch.size: 500
        23
        canal.tcp.username:
        24
        canal.tcp.password:
        25
        # kafka consumer
        26
        kafka.bootstrap.servers: 127.0.0.1:9092
        27
        kafka.enable.auto.commit: false
        28
        kafka.auto.commit.interval.ms: 1000
        29
        kafka.auto.offset.reset: latest
        30
        kafka.request.timeout.ms: 40000
        31
        kafka.session.timeout.ms: 30000
        32
        kafka.isolation.level: read_committed
        33
        kafka.max.poll.records: 1000
        34
        # rocketMQ consumer
        35
        rocketmq.namespace:
        36
        rocketmq.namesrv.addr: 127.0.0.1:9876
        37
        rocketmq.batch.size: 1000
        38
        rocketmq.enable.message.trace: false
        39
        rocketmq.customized.trace.topic:
        40
        rocketmq.access.channel:
        41
        rocketmq.subscribe.filter:
        42
        # rabbitMQ consumer
        43
        rabbitmq.host:
        44
        rabbitmq.virtual.host:
        45
        rabbitmq.username:
        46
        rabbitmq.password:
        47
        rabbitmq.resource.ownerId:
        48
        ​
        49
        srcDataSources:
        50
        defaultDS:
        51
        url: jdbc:mysql://127.0.0.1:3306/msb_order?useUnicode=true
        52
        username: root
        53
        password: 123456
        54
        canalAdapters:
        55
        - instance: example # canal instance Name or mq topic name
        56
        groups:
        57
        - groupId: g1
        58
        outerAdapters:
        59
        - name: logger
        60
        - name: es7
        61
        hosts: 127.0.0.1:9200 # 127.0.0.1:9200 for rest mode
        62
        key: exampleKey
        63
        properties:
        64
        mode: rest # transport or rest
        65
        # security.auth: test:123456 # only used for rest mode
        66
        cluster.name: elasticsearch
        Copied!
        • 配置conf/es7/my_order.yml
        1
        dataSourceKey: defaultDS # 源数据源的key, 对应上面配置的srcDataSources中的值
        2
        outerAdapterKey: exampleKey # 对应application.yml中es配置的key
        3
        destination: example # cannal的instance或者MQ的topic
        4
        groupId: g1 # 对应MQ模式下的groupId, 只会同步对应groupId的数据_search
        5
        esMapping:
        6
        _index: msb_order # es 的索引名称
        7
        _type: _doc # es7 固定'_doc'，es8删除
        8
        _id: _id # es 的_id, 如果不配置该项必须配置下面的pk项_id则会由es自动分配
        9
        upsert: true
        10
        sql: "SELECT a.id as _id, a.customer_phone, a.customer_name, a.customer_region, a.customer_city, a.customer_district, a.customer_addr, a.failure_phenomenon, a.failure_phenomenon_text, a.book_date, a.service_required, a.buyshop_detail, a.buy_shop, a.buy_way, a.product_type, a.model_no, a.process_type, a.ex_order_no, a.customer_country, a.source, a.matnr, a.buyDate, a.serviceReq, a.create_time, a.result, a.rvstatus, a.ip_address FROM msb_order_1 a"
        11
        commitBatch: 3000
        Copied!
        • 服务启动
        • 注意：
          • 索引msb_order的mapping必须提前创建好
          • 索引中的id字段是_id，因此需要查询的时候需要id as _id
      7. 7.
        按时

搜索推荐词

Query Suggestion
​
​

过滤

filter 对结果进行过滤。filter 不会贡献相关性得分。
Previous
MySQL-103
Next
Logstash
Last modified 2mo ago
Copy link
Contents
技术栈
视频教程
数据高可用
index, shards, replicas
基于canal的MySQL=>ES数据同步方案
搜索推荐词
过滤