otter自定义数据同步踩坑实录

亦凉 2024-04-18 16:36 208阅读 0赞

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][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文本里面粘贴所有代码。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RyYWdvbnBlbmcyMDA4_size_16_color_FFFFFF_t_70][]

3、增加字段不生效问题解决

数据过滤和字典转换能够正常执行，但是增加字段一直不生效，也不报错，查了好久找到了解决方法。

在github源码的issues里面找到了答案

地址：[https://github.com/alibaba/otter/issues/731][https_github.com_alibaba_otter_issues_731]

新字段需要在映射关系表中定义（源、目标表中都需要配置），ViewExtractor.java 中有过滤 如果新字段源表没有，目标表有，需要扩展下 ViewExtractor

需要在otter.COLUMN\_PAIR表里为新增加的字段添加映射关系，否则会被过滤掉。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RyYWdvbnBlbmcyMDA4_size_16_color_FFFFFF_t_70 1][]

另外，在issues上看到有人做了一个自动化批量配置的工具，可以看下

[https://github.com/dongjiashun/otter\_auto\_tools][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管理，点击编辑，点击高级设置，将启用数据表类型转化选择“是”

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RyYWdvbnBlbmcyMDA4_size_16_color_FFFFFF_t_70 2][]

参考：[https://blog.csdn.net/u013705066/article/details/86471149][https_blog.csdn.net_u013705066_article_details_86471149]

[https_github.com_alibaba_otter_wiki_E6_98_A0_E5_B0_84_E8_A7_84_E5_88_99_E9_85_8D_E7_BD_AE]: https://github.com/alibaba/otter/wiki/%E6%98%A0%E5%B0%84%E8%A7%84%E5%88%99%E9%85%8D%E7%BD%AE
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RyYWdvbnBlbmcyMDA4_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/0a77afd96e4b49f6aab7ed23e9efc860.png
[https_github.com_alibaba_otter_issues_731]: https://github.com/alibaba/otter/issues/731
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RyYWdvbnBlbmcyMDA4_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/6c3edf0930354c3198e1aa67063c2893.png
[https_github.com_dongjiashun_otter_auto_tools]: https://github.com/dongjiashun/otter_auto_tools
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RyYWdvbnBlbmcyMDA4_size_16_color_FFFFFF_t_70 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/a0f550de745f4f7b9d1242093ecf0dc1.png
[https_blog.csdn.net_u013705066_article_details_86471149]: https://blog.csdn.net/u013705066/article/details/86471149