目的：將資料庫DBA中的a表,實時同步到DBB庫中的b表，b表與a表相比多了一個新的欄位進行區分；

部署：

部署zk;

新增zk/node/canal到manager控制；

新增資料來源、資料表管理

新增channel/pipeline/對映關係

eventProcessor:

 package com.alibaba.otter.node.extend.processor;
import java.sql.Types;
import java.util.List;


import com.alibaba.otter.node.extend.processor.AbstractEventProcessor;
import com.alibaba.otter.shared.etl.model.EventColumn;
import com.alibaba.otter.shared.etl.model.EventData;
import com.alibaba.otter.shared.etl.model.EventType;
/**
 * 業務自定義處理過程
 *
 * @author jianghang 2012-6-25 下午02:26:36
 * @version 4.1.0
 */
public class TestTranfer extends AbstractEventProcessor {


    /**
     * 自定義處理單條EventData物件
     *
     * @return {@link EventData} 返回值=null，需要忽略該條資料
     */
    public boolean process(EventData eventData){
             EventType et = eventData.getEventType();
        if(et == EventType.INSERT){
            insert(eventData);
        }
        
        if(et == EventType.UPDATE || et == EventType.DELETE){
            EventColumn misc = new EventColumn();
            misc.setColumnValue("122");
            misc.setColumnType(Types.INTEGER);
            misc.setColumnName("addli");
            List<EventColumn> ks = eventData.getKeys();
            ks.add(misc);
        }
        return true;
    }


    private void printList(List<EventColumn> list){
        for(EventColumn t:list){
            System.out.println(t.getColumnName() +" = " + t.getColumnValue());
        }
    }
    
    private void insert(EventData eventData) {
        EventColumn misc = new EventColumn();
        misc.setColumnValue("122");
        misc.setColumnType(Types.INTEGER);
        misc.setColumnName("addli");
        misc.setUpdate(true);
        List<EventColumn> cols = eventData.getUpdatedColumns();
        cols.add(misc);
        eventData.setColumns(cols);
    }
} 
 

測試用資料庫指令碼：

CREATE DATABASE /*!32312 IF NOT EXISTS*/`mycat_test` /*!40100 DEFAULT CHARACTER SET utf8 */;  
  
USE `mycat_test`;  
  
/*Table structure for table `sam_test` */  
  
DROP TABLE IF EXISTS `sam_test`;  
  
CREATE TABLE `sam_test` (  
  `id_` int(4) NOT NULL AUTO_INCREMENT,  
  `name_` varchar(64) NOT NULL,  
  `user_id` bigint(4) NOT NULL,  
  PRIMARY KEY (`id_`)  
) ENGINE=InnoDB AUTO_INCREMENT=4 DEFAULT CHARSET=utf8;  
 

CREATE DATABASE /*!32312 IF NOT EXISTS*/`mycat_test1` /*!40100 DEFAULT CHARACTER SET utf8 */;  
  
USE `mycat_test1`;  
  
/*Table structure for table `sam_test1` */  
  
DROP TABLE IF EXISTS `sam_test1`;  
  
CREATE TABLE `sam_test1` (  
  `id_` int(4) NOT NULL AUTO_INCREMENT,  
  `name_` varchar(64) NOT NULL,  
  `user_id` bigint(4) NOT NULL, `addli` int(6) not null, 
  PRIMARY KEY (`id_`)  
) ENGINE=InnoDB AUTO_INCREMENT=4 DEFAULT CHARSET=utf8;