美文网首页
04.记一次基于otter实现数据加工转换,开发自定义数据转换的

04.记一次基于otter实现数据加工转换,开发自定义数据转换的

作者: 乐活骑士 | 来源:发表于2020-07-26 11:41 被阅读0次

    什么是otter

    附上官方介绍,感兴趣的可以去学习。原文地址:项目介绍

    image.png

    otter与canal

    otter与canal

    1.jpeg

    正文开始:基于otter开发自定义数据转换

            目前我们有使用otter进行数据同步,但是我们也挖掘了otter中ETL转化、数据异构的功能。具体做法请往下看。欢迎大家一起讨论。

    如何把大象放进冰箱

    1. 设计异构业务表
    2. otter配置同步
    3. 开发自定义事件处理器
    4. 上传配置

    01.设计异构业务表

            首先设计出你需要异构的表结构。

    02.otter配置同步

            在otter后台配置源表和目标表映射。


    image-1.png image-2.png

    03.开发自定义事件处理器

            从git上拉取otter源码otter,切换到指定分支或者tag。继承com.alibaba.otter.node.extend.processor.AbstractEventProcessor,同时实现com.alibaba.otter.shared.etl.extend.processor.support.DataSourceFetcherAware接口。

    image-3.png

            好了,现在就可以进行具体业务开发了。代码骨架如下:

    package com.alibaba.otter.node.extend.processor.demo;
    
    import com.alibaba.otter.node.extend.processor.AbstractEventProcessor;
    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 com.alibaba.otter.shared.etl.model.EventType;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.jdbc.core.JdbcTemplate;
    import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
    
    import javax.sql.DataSource;
    import java.sql.Types;
    import java.util.ArrayList;
    import java.util.List;
    import java.util.Map;
    import java.util.stream.Collectors;
    
    /**
     * demo
     * 注意:
     *  1、写sql查询时,表名前要带上schema
     *  2、多写点try-catch,避免执行任务挂起
     *
     * @author William Zhang
     * @return
     * @since 2020/7/22 8:57 下午
     */
    public class GuideEventProcessor extends AbstractEventProcessor implements DataSourceFetcherAware {
    
        private Logger logger = LoggerFactory.getLogger(GuideEventProcessor.class);
        // TODO jdbcTemplate二者取一即可
        private JdbcTemplate jdbcTemplate = new JdbcTemplate();
        private NamedParameterJdbcTemplate namedParameterJdbcTemplate = null;
    
        /**
         * 设置数据源信息
         *
         * @param dataSourceFetcher
         * @return void
         * @author William Zhang
         * @since 2020/7/24 3:11 下午
         */
        @Override
        public void setDataSourceFetcher(DataSourceFetcher dataSourceFetcher) {
            // TODO 这个Id,取otter配置后所分配的"数据表序号"。访问 http://xxx:8080/dataMediaInfo.htm?dataMediaId=113 可以看到。
            DataSource dataSource = dataSourceFetcher.fetch(115L);
            jdbcTemplate.setDataSource(dataSource);
            namedParameterJdbcTemplate = new NamedParameterJdbcTemplate(dataSource);
        }
    
        /**
         * 这是处理业务的方法 TODO
         *
         * @param eventData
         * @return boolean
         * @author William Zhang
         * @since 2020/7/24 3:09 下午
         */
        @Override
        public boolean process(EventData eventData) {
            try {
                // ==> 准备数据
                // 表名
                String tableName = eventData.getTableName();
                // 操作字段集合
                List<EventColumn> cols = new ArrayList<>();
                // 操作字段主键集合(如果是联合主键,bigTable上也要设置成联合主键,或者联合唯一约束,不然不生效)
                List<EventColumn> keys = new ArrayList<>();
                // 原始字段名-字段信息映射
                Map<String, EventColumn> colNameColMap = eventData.getColumns().stream().collect(Collectors.toMap(EventColumn::getColumnName, c -> c));
    
                // ==> 替换为新的字段和主键信息
                // TODO
                cols.add(this.getNewColumn("col1", "lehuo", Types.VARCHAR));
                cols.add(this.getNewColumn("col2", "168", Types.INTEGER));
                cols.add(this.getNewColumn("col3", "2020-07-26 11:46:40", Types.TIMESTAMP));
                eventData.setColumns(cols);
    
                // ==> 构造新的主键,设置联合主键的话,一定要在表中配置成联合主键。
                keys.add(this.getNewColumn("xxx1", "xxx1", Types.VARCHAR));
                keys.add(this.getNewColumn("xxx2", "xxx2", Types.VARCHAR));
                eventData.setKeys(keys);
    
                // ==> 修改数据meta信息。如果已存在,会执行update
                eventData.setEventType(EventType.INSERT);
                eventData.setSchemaName("du_wms");
                eventData.setTableName("report_received_shipment");
            } catch (Exception e) {
                logger.error("处理数据时出现异常", e.getMessage(), e);
                return false;
            }
            return true;
        }
    
    
        /**
         * @param colName
         * @param colType  Types.VARCHAR
         * @param colValue
         */
        private EventColumn getNewColumn(String colName, String colValue, int colType) {
            EventColumn taskNoCol = new EventColumn();
            taskNoCol.setColumnName(colName);
            taskNoCol.setColumnValue(colValue);
            taskNoCol.setColumnType(colType);
            return taskNoCol;
        }
    
    
    }
    
    image-4.png image-5.png

    04.上传配置

            现在,你已经开发好了你的业务,该让它生效了,步骤如下:

    STEP

    1. 登录otter管理后台

      image-6.png
    2. 停用channel,停用后方可编辑

      image-7.png
    3. 编辑pipeline下的映射

      image-8.png
    4. 上传刚才编写的自定义EventProcessor代码,如图。

      image-9.png
    5. 启动channel

      image-10.png

            完成以上配置后,就可以工作了(当然要保证自定义EventProcessor的逻辑正确)。

    调试

            otter无法进行本地调试,调试的话需要启动远程调试模式。
    sh start.sh debug 5005
    然后再Idea中设置Debug->Remote->填入ip和端口号->apply,就可以开始远程调试。如图。

    image-11.png

    参考资料

    otter官方文档
    常见问题
    自定义数据同步

    相关文章

      网友评论

          本文标题:04.记一次基于otter实现数据加工转换,开发自定义数据转换的

          本文链接:https://www.haomeiwen.com/subject/lmdwlktx.html