什么是otter

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

image.png

otter与canal

1.jpeg

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

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

如何把大象放进冰箱

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

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;
    }


}