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;
}
}
场景二：自定义数据处理逻辑，包括读库操作(实现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));
}
}
————————————————
扩展性定义
按照实现不同，可分为两类：
数据处理自定义，比如Extract , Transform的数据处理. 目前Select/Load不支持数据自定义处理
组件功能性扩展，比如支持oracle日志获取，支持hbase数据输出等.
数据处理自定义
Extract模块：
EventProcessor : 自定义数据处理，可以改变一条变更数据的任意内容
FileResolver : 解决数据和文件的关联关系
目前两者都只支持java语言编写，但都支持运行时动态编译&lib包载入的功能。
通过Otter Manager直接发布source文件代码，然后推送到node节点上即时生效，不需要重启任何java进程，有点动态语言的味道
可以将class文件放置到extend目录或者打成jar包，放置在node启动classpath中，也可以通过Otter Manager指定类名的方式进行加载，这样允许业务完全自定义。(但有个缺点，如果使用了一些外部包加入到node classpath中，比如远程接口调用，目前EventProcessor的调用是串行处理，针对串行进行远程调用执行，效率会比较差. 如有并行处理需求，请联系我，可增加此功能)
EventProcessor接口示例：
/**
* 业务自定义处理过程
*
* @author jianghang 2012-6-25 下午02:26:36
* @version 4.1.0
*/
public interface EventProcessor {
/**
* 自定义处理单条EventData对象
*
* @return {@link EventData} 返回值=null，需要忽略该条数据
*/
public EventData process(EventData eventData);
}
组件功能性扩展
目前这块扩展性机制不够，设计时只预留了接口，但新增一个功能实现，需要通过硬编码的方式去进行，下载otter的源码，增加功能支持，修改spring配置，同时修改web页面，方便使用。
基于manager的灵活扩展性的实现，暂没有想到很好的办法，如果你有好的思路和实现方式，也可以告知我们，谢谢。
比如举增加hbase load实现为例，需要扩展的内容：
增加hbase数据源的抽象
增加hbase表的抽象，比如column，columnFamily
增加hbase transform的实现
增加hbase loader的实现
https://www.it610.com/article/2557050.htm
https://blog.csdn.net/u013705066/article/details/86471149
