MySql 数据同步（同步关系型、非关系型(hadoop)）--canal

矫情吗；* 2022-06-01 11:57 169阅读 0赞

参考博客：[http://blog.csdn.net/hackerwin7/article/details/37923607][http_blog.csdn.net_hackerwin7_article_details_37923607]

**注意**：本次只是实时获取到了mysql的同步日志 你可以封装成json格式 扔到消息队列 消费这些数据 对hadoop进行更新。

我用的消息队列是kafka 执行的是HBase

首先先说canal的配置：  
[https://github.com/alibaba/canal/releases][https_github.com_alibaba_canal_releases]下载最新版本的canal.deployer,解压  
下载下来 四个目录 bin conf lib logs

**mysql的改动**  
先修改 mysql的my.cnf  
（1）开启mysql的binlog功能，并配置binlog模式为row。  
在my.cnf 加入如下：

[mysqld]  
    log-bin=mysql-bin #添加这一行就ok 
    binlog-format=ROW #选择row模式 
    server_id=1 #配置mysql replaction需要定义，不能和canal的slaveId重复

mysql创建用户 授权

CREATE USER canal IDENTIFIED BY 'canal';    
    GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';  
    -- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ; 
    FLUSH PRIVILEGES;

**canal的改动**  
修改两处 1.conf下面的canal.properties

canal.id= 1
    canal.ip=
    canal.port= 11111
    canal.zkServers=
    # flush data to zk
    canal.zookeeper.flush.period = 1000
    # flush meta cursor/parse position to file
    canal.file.data.dir = ${canal.conf.dir}
    canal.file.flush.period = 1000
    ## memory store RingBuffer size, should be Math.pow(2,n)
    canal.instance.memory.buffer.size = 32768
    ## memory store RingBuffer used memory unit size , default 1kb
    canal.instance.memory.buffer.memunit = 1024 
    ## meory store gets mode used MEMSIZE or ITEMSIZE
    canal.instance.memory.batch.mode = MEMSIZE

2.修改 conf/example下面的instance.properties

#################################################
    ## mysql serverId
    canal.instance.mysql.slaveId=1234   
    # position info
    canal.instance.master.address=127.0.0.1:3306  //mysql数据库位置
    canal.instance.master.journal.name=
    canal.instance.master.position=
    canal.instance.master.timestamp=
    
    
    # table meta tsdb info
    canal.instance.tsdb.enable=true
    canal.instance.tsdb.dir=${canal.file.data.dir:../conf}/${canal.instance.destination:}
    canal.instance.tsdb.url=jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;
    #canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
    canal.instance.tsdb.dbUsername=canal
    canal.instance.tsdb.dbPassword=canal
    
    
    #canal.instance.standby.address =
    #canal.instance.standby.journal.name =
    #canal.instance.standby.position = 
    #canal.instance.standby.timestamp = 
    # username/password 
    /*mysql创建的用户密码 默认库 连接字符串*/
    canal.instance.dbUsername=canal
    canal.instance.dbPassword=canal 
    canal.instance.defaultDatabaseName=e_mall
    canal.instance.connectionCharset=UTF-8
    # table regex
    canal.instance.filter.regex=.*\\..*
    # table black regex
    canal.instance.filter.black.regex=

然后启动canal  
查看启动日志：  
vim canal/log/canal/canal.log  
如果出现 ：

2014-07-18 10:21:08.525 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## start the canal server. 
    2014-07-18 10:21:08.609 [main] INFO  com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[10.12.109.201:11111] 
    2014-07-18 10:21:09.037 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## the canal server is running now ......

说明启动成功

下面是测试方法 ：  
运行下面的代码 监听 mysql 更新一条数据 如果有数据过来说明成功

maven:

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

package log.binary;
    
    import java.net.InetSocketAddress;
    import java.util.List;
    
    import com.alibaba.otter.canal.client.CanalConnector;
    import com.alibaba.otter.canal.common.utils.AddressUtils;
    import com.alibaba.otter.canal.protocol.Message;
    import com.alibaba.otter.canal.protocol.CanalEntry.Column;
    import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
    import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
    import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
    import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
    import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
    import com.alibaba.otter.canal.client.*;
    import org.jetbrains.annotations.NotNull;
    
    public class ClientSample {
    
        public static void main(String args[]) {
            // 创建链接
            CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.*.*",
                    11111), "example", "", "");
            int batchSize = 1000;
            int emptyCount = 0;
            try {
                connector.connect();
                connector.subscribe(".*\\..*");
                connector.rollback();
                int totalEmtryCount = 1200;
                while (emptyCount < totalEmtryCount) {
                    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) {
                            e.printStackTrace();
                        }
                    } 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(@NotNull 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(@NotNull List<Column> columns) {
            for (Column column : columns) {
                System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated());
            }
        }
    }

[http_blog.csdn.net_hackerwin7_article_details_37923607]: http://blog.csdn.net/hackerwin7/article/details/37923607
[https_github.com_alibaba_canal_releases]: https://github.com/alibaba/canal/releases