ElasticSearch - 深度学习 - Canal

╰+哭是因爲堅強的太久メ 2023-02-20 06:28 1阅读 0赞

## 前言 ##

整理一下ElasticSearch中使用的场景

## 索引更新 ##

索引构建主要分为两种

*  全量索引构建：从头全部重新建，通常因为：首次构建、增减字段、分词修改、时间久了增量丢数据后
 *  增量索引构建：只创建或者修改更新的数据，有如下优势
    
     *  准确性高：数据变更后，es马上更新，不影响用户体验
     *  性能快：电商高并发场景下
     *  高可用、实现简单：es集群、canal

## 增量构建索引的可选方案 ##

\*\*单系统应用：\*\*比较简单，插入数据库时更新es数据，如失败进行补偿机制

\*\*分布式系统：\*\*搜索中心作为独立部署，大部分情况下是这样的

*  利用成熟的MQ中间件：通常的做法就是业务系统有数据变更时发送Message，搜索中心监听消费消息
 *  通过SQL查询语句定时扫描数据库：轮询进行更新，但是：DELETE、数据量大的情况下 都会出问题
 *  阿里巴巴canal中间件：原理通过数据库bin\_log 进行差异化更新

## 使用Canal ##

Github官方：https://github.com/alibaba/canal

### canal 工作原理 ###

*  canal 模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave ，向 MySQL master 发送dump 协议
 *  MySQL master 收到 dump 请求，开始推送 binary log 给 slave (即 canal )
 *  canal 解析 binary log 对象(原始为 byte 流)

### MySQL 配置 ###

mysql.ini `其实跟标准主从配置没什么区别，主要是 写入格式为 ROW`

[mysqld]
    # 开启 binlog
    log-bin=mysql-bin 
    # 选择 ROW 模式
    binlog-format=ROW 
    # 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复
    server_id=1

canal-server在启动时，会去dump表结构并缓存，要求添加\`\`权限

CREATE USER canal IDENTIFIED BY 'canal';  
    # 为了先跑通，先全开一会回收
    GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
    FLUSH PRIVILEGES;

授权 canal 链接 MySQL 账号具有作为 MySQL slave 的权限, 如果已有账户可直接 grant

REVOKE ALL PRIVILEGES ON *.* FROM 'canal'@'%' ;
    GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
    FLUSH PRIVILEGES;

查看

show grants for 'canal'@'%';

### Docker快速入门 ###

docker pull canal/canal-server:v1.1.1

启动

docker run -d \
    -p 2222:2222 \
    -p 8000:8000 \
    -p 11111:11111 \
    -p 11112:11112 \
    --name my-canal \
    -e canal.auto.scan=false \
    		  -e canal.destinations=test \
    		  -e canal.instance.master.address=host.docker.internal:3306  \
    		  -e canal.instance.dbUsername=canal  \
    		  -e canal.instance.dbPassword=canal  \
    		  -e canal.instance.connectionCharset=UTF-8 \
    		  -e canal.instance.tsdb.enable=true \
    		  -e canal.instance.gtidon=false \
    canal/canal-server:v1.1.1

如果有问题，可以查看日志

docker exec -it my-canal tail -f /home/admin/canal-server/logs/canal/canal.log

### Java 配置 ###

引入依赖

<dependency>
        <groupId>com.alibaba.otter</groupId>
        <artifactId>canal.client</artifactId>
        <version>1.1.4</version>
    </dependency>

代码

public class SimpleCanalClientExample { 
    
        public static void main(String args[]) { 
            CanalConnector connector = CanalConnectors.newSingleConnector(
                    new InetSocketAddress("127.0.0.1", 11111), "test", "canal", "canal");
            int batchSize = 1000;
            int emptyCount = 0;
            try { 
                connector.connect();
                connector.subscribe(".*\\..*");
                connector.rollback();
                int totalEmptyCount = 120;
                while (emptyCount < totalEmptyCount) { 
                    // 获取指定数量的数据
                    Message message = connector.getWithoutAck(batchSize); 
                    long batchId = message.getId();
                    int size = message.getEntries().size();
                    if (batchId == -1 || size == 0) { 
                        emptyCount++;
                        System.out.println("empty count : " + emptyCount);
                        try { 
                            Thread.sleep(1000);
                        } catch (InterruptedException e) { 
                        }
                    } else { 
                        emptyCount = 0;
                        // System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
                        printEntry(message.getEntries());
                    }
    
                    connector.ack(batchId); // 提交确认
                    // connector.rollback(batchId); // 处理失败, 回滚数据
                }
    
                System.out.println("empty too many times, exit");
            } finally { 
                connector.disconnect();
            }
        }
    
        private static void printEntry(List<Entry> entrys) { 
            for (Entry entry : entrys) { 
                if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) { 
                    continue;
                }
    
                RowChange rowChage = null;
                try { 
                    rowChage = RowChange.parseFrom(entry.getStoreValue());
                } catch (Exception e) { 
                    throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                            e);
                }
    
                EventType eventType = rowChage.getEventType();
                System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
                        entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                        entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                        eventType));
    
                for (RowData rowData : rowChage.getRowDatasList()) { 
                    if (eventType == EventType.DELETE) { 
                        printColumn(rowData.getBeforeColumnsList());
                    } else if (eventType == EventType.INSERT) { 
                        printColumn(rowData.getAfterColumnsList());
                    } else { 
                        System.out.println("-------> before");
                        printColumn(rowData.getBeforeColumnsList());
                        System.out.println("-------> after");
                        printColumn(rowData.getAfterColumnsList());
                    }
                }
            }
        }
    
        private static void printColumn(List<Column> columns) { 
            for (Column column : columns) { 
                System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated());
            }
        }
    
    }

## 测试 ##

创建测试表

CREATE TABLE `xdual` (
    `ID` int(11) NOT NULL AUTO_INCREMENT,
    `X` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
    PRIMARY KEY (`ID`)
    ) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8

insert into xdual(id,x) values(null,now());

## 错误解决 ##

https://blog.csdn.net/chenzuancong9642/article/details/101040754