otter自定义数据同步
otter支持数据处理自定义过程。
Extract模块：
EventProcessor : 自定义数据处理，可以改变一条变更数据的任意内容
FileResolver : 解决数据和文件的关联关系
目前两者都只支持java语言编写，但都支持运行时动态编译&lib包载入的功能。
通过Otter Manager直接发布source文件代码，然后推送到node节点上即时生效，不需要重启任何java进程，有点动态语言的味道
可以将class文件放置到extend目录或者打成jar包，放置在node启动classpath中，也可以通过Otter Manager指定类名的方式进行加载，这样允许业务完全自定义。(但有个缺点，如果使用了一些外部包加入到node classpath中，比如远程接口调用，目前EventProcessor的调用是串行处理，针对串行进行远程调用执行，效率会比较差. )
官方具体说明参加：https://github.com/alibaba/otter/wiki/%E6%98%A0%E5%B0%84%E8%A7%84%E5%88%99%E9%85%8D%E7%BD%AE
1、编写自定义数据同步代码
首先在node-->extend路径下编写自定义扩展类UserTestEventProcessor，需要继承AbstractEventProcessor接口
package com.alibaba.otter.node.extend.processor;
import com.alibaba.otter.shared.etl.model.EventColumn;
import com.alibaba.otter.shared.etl.model.EventData;
import com.alibaba.otter.shared.etl.model.EventType;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.sql.Types;
import java.util.Date;
import java.util.List;
public class UserTestEventProcessor extends AbstractEventProcessor {
public static final Logger log = LoggerFactory.getLogger("UserTestEventProcessor");
@Override
public boolean process(EventData eventData) {
if (StringUtils.equals("user_0", eventData.getTableName())) {
// EventColumn eventColumn = getColumn(eventData , "create_user");
for (EventColumn column : eventData.getColumns()) {
//如果create_user字段值为3，则不同步该记录
if (column.getColumnName().equals("create_user")) {
if (column.getColumnValue().equals("3")) {
return false;
}
//改字段名，该字段就不会被处理
//column.setColumnName("update_user");
}
//字典值转换
else if (column.getColumnName().equals("dict_type")) {
if (column.getColumnValue().equals("1")) {
column.setColumnValue("1004001");
} else {
column.setColumnValue("1004002");
}
} else {
}
}
//注意，新字段需要在映射关系表(otter.COLUMN_PAIR)中定义（源、目标表中都需要配置），ViewExtractor.java 中有过滤
//如果新字段源表没有，目标表有，需要扩展下 ViewExtractor
//增加字段update_time1，值取create_time字段的值
EventColumn timeColumn = getColumn(eventData, "create_time");
EventColumn updateTime1 = timeColumn.clone();
//timestamp,Datetime是一个long型的数字.
updateTime1.setColumnName("update_time1");
//updateTime1.setIndex(7);
//设置该字段为更新，否则不会执行同步
updateTime1.setUpdate(true);
List<EventColumn> cols = eventData.getColumns();
cols.add(updateTime1);
eventData.setColumns(cols);
printList(cols);
//增加字段addli
EventColumn misc = new EventColumn();
misc.setColumnValue("122");
misc.setColumnType(Types.VARCHAR);
misc.setColumnName("addli");
//misc.setIndex(6);
misc.setUpdate(true);
eventData.getColumns().add(misc);
log.error("!========misc:" + misc.toString());
List<EventColumn> cols1 = eventData.getColumns();
//printList(cols1);
}
return true;
}
private void printList(List<EventColumn> list) {
for (EventColumn t : list) {
log.error("!!!!!!!!!==============" + t.toString());
}
}
}
2、配置代码
然后将该类的所有代码拷贝到
Channel管理 > Pipeline管理 > 映射关系管理 > 编辑映射关系
EventProcessor类型选择Source，然后在EventProcessor文本里面粘贴所有代码。
3、增加字段不生效问题解决
数据过滤和字典转换能够正常执行，但是增加字段一直不生效，也不报错，查了好久找到了解决方法。
在github源码的issues里面找到了答案
地址：https://github.com/alibaba/otter/issues/731
新字段需要在映射关系表中定义（源、目标表中都需要配置），ViewExtractor.java 中有过滤 如果新字段源表没有，目标表有，需要扩展下 ViewExtractor
需要在otter.COLUMN_PAIR表里为新增加的字段添加映射关系，否则会被过滤掉。
另外，在issues上看到有人做了一个自动化批量配置的工具，可以看下
https://github.com/dongjiashun/otter_auto_tools
4、数据处理扩展的其他示例代码
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));
}
}
5、invalid character异常解决
同步过程中发现报invalid character，下面错误：
Caused by: java.util.concurrent.ExecutionException: com.alibaba.otter.node.etl.load.exception.LoadException: com.alibaba.otter.node.etl.load.exception.LoadException: org.springframework.jdbc.BadSqlGrammarException: StatementCallback; bad SQL grammar []; nested exception is java.sql.SQLSyntaxErrorException: ORA-00911: invalid character at java.util.concurrent.FutureTask.report(FutureTask.java:122) at java.util.concurrent.FutureTask.get(FutureTask.java:192) at com.alibaba.otter.node.etl.load.loader.db.DataBatchLoader.load(DataBatchLoader.java:107) at com.alibaba.otter.node.etl.load.loader.OtterLoaderFactory.load(OtterLoaderFactory.java:50) at com.alibaba.otter.node.etl.load.LoadTask$1.run(LoadTask.java:85) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748)
解决方法：打开Channel管理--Pipeline管理，点击编辑，点击高级设置，将启用数据表类型转化选择“是”
参考：https://blog.csdn.net/u013705066/article/details/86471149