otter自定义扩展

otter自定义扩展

otter支持数据处理自定义过程。
Extract模块：

EventProcessor : 自定义数据处理，可以改变一条变更数据的任意内容
FileResolver : 解决数据和文件的关联关系
目前两者都只支持java语言编写，但都支持运行时动态编译&lib包载入的功能。

通过Otter Manager直接发布source文件代码，然后推送到node节点上即时生效，不需要重启任何java进程，有点动态语言的味道
可以将class文件放置到extend目录或者打成jar包，放置在node启动classpath中，也可以通过Otter Manager指定类名的方式进行加载，这样允许业务完全自定义。(但有个缺点，如果使用了一些外部包加入到node classpath中，比如远程接口调用，目前EventProcessor的调用是串行处理，针对串行进行远程调用执行，效率会比较差. )

数据处理扩展的示例代码

场景一：根据业务逻辑判断是否同步该条数据

package com.alibaba.otter.node.extend.processor;

import com.alibaba.otter.shared.etl.model.EventColumn;
import com.alibaba.otter.shared.etl.model.EventData;
import org.apache.commons.lang.StringUtils;


public class TestEventProcessor extends AbstractEventProcessor {

    public boolean process(EventData eventData) {

        boolean isHandle = true;

        if(StringUtils.equals("test_all", eventData.getTableName())){
            EventColumn eventColumn = getColumn(eventData , "type");

            //如果类型是2的话则忽略
            //可以类比 如果不是需要同步的数据，则忽略同步
            if("2".equals(eventColumn.getColumnValue())){
                isHandle = false;
            }
        }

        return isHandle;

    }


}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30

场景二：自定义数据处理逻辑，包括读库操作(实现DataSourceFetcherAware接口)

package com.alibaba.otter.node.extend.processor;

import com.alibaba.otter.shared.etl.extend.processor.support.DataSourceFetcher;
import com.alibaba.otter.shared.etl.extend.processor.support.DataSourceFetcherAware;
import com.alibaba.otter.shared.etl.model.EventColumn;
import com.alibaba.otter.shared.etl.model.EventData;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;


public class TestEventProcessor extends AbstractEventProcessor implements DataSourceFetcherAware {

    private final static Logger logger = LoggerFactory.getLogger(TestEventProcessor.class);

    protected DataSource dataSource;

    public boolean process(EventData eventData) {

        boolean isHandle = true;

        if(StringUtils.equals("test_all", eventData.getTableName())){
            EventColumn typeColumn = getColumn(eventData , "type");
            //如果类型是2的话则忽略
            //可以类比 如果不是需要同步的数据，则忽略同步
            if("2".equals(typeColumn.getColumnValue())){
                isHandle = false;
            }else {
                try {
                    Connection conn = this.dataSource
                            .getConnection();

                    //去操作该数据库里的其他表获取关联字段
                    ResultSet set =conn.prepareStatement("SELECT * FROM `test`.`main` WHERE 1")
                            .executeQuery();

                    //自定义修改数据
                    if(set.next()){
                        EventColumn titleColumn = getColumn(eventData , "title");
                        titleColumn.setColumnValue(titleColumn.getColumnValue() + set.getString("name"));
                    }


                    //释放资源
                    set.close();
                    conn.close();

                } catch (SQLException e) {
                    logger.error(e.getMessage());
                }

            }

        }

        return isHandle;

    }


    @Override
    public void setDataSourceFetcher(DataSourceFetcher dataSourceFetcher) {
        //这里是表的id,也就是源数据库里的某个表
        this.dataSource = dataSourceFetcher.fetch(new Long(3));
    }
}