简介

反向etl区别与etl,其主要目的在于数据仓库到下游系统, etl更多的是为了 数仓的创建.
反向etl目的在于提供一个标准声明,方便下游app共享基础能力和扩展.
用途: 主数据订阅/第三方数据订阅等

核心组件

  • easybatch 批处理
  • xxljob 分布式调度
  • pf4j 插件化开发
  • dremio 数据湖(大于数仓,提供更强大的olap能力)

easybatch 批处理结合 xxljob调度即可实现 简单可靠的数据分发功能,配合数据湖/pf4j 插件开发,可以实现相当灵活的 : 反向etl功能/数据订阅.

  1. public interface IBatch<SOURCE, SINK> {
  3.     JobExecutor jobExecutor();
  5.     RecordReader reader(SOURCE i);
  7.     RecordWriter writer(BatchResult result, SINK o);
  9.     RecordMapper mapper();
  11.     default CustomPipelineListener customPipelineListener(String jobId) {
  12.         return new CustomPipelineListener(jobId);
  13.     }
  15.     default CustomWriterListener customWriterListener(String jobId) {
  16.         return new CustomWriterListener(jobId);
  17.     }
  19.     default CustomReaderListener customReaderListener(String jobId) {
  20.         return new CustomReaderListener(jobId);
  21.     }
  23.     // 可选
  24.     default RecordProcessor processor() {
  25.         return null;
  26.     }
  28.     default BatchResult execute(SOURCE source, SINK sink) {
  29.         BatchResult result = new BatchResult(UUID.randomUUID().toString(), XxlJobHelper.getJobId());
  30.         JobExecutor jobExecutor = jobExecutor();
  31.         JobBuilder jobBuilder = new JobBuilder();
  32.         jobBuilder.reader(reader(source));
  33.         jobBuilder.writer(writer(result, sink));
  34.         //jobBuilder.batchSize(100);
  35.         RecordMapper mapper = mapper();
  36.         if (mapper != null) {
  37.             jobBuilder.mapper(mapper);
  38.         }
  39.         RecordProcessor processor = processor();
  40.         if (processor != null) {
  41.             jobBuilder.processor(processor);
  42.         }
  43.         jobBuilder.readerListener(customReaderListener(result.getJobId()));
  44.         jobBuilder.writerListener(customWriterListener(result.getJobId()));
  45.         jobBuilder.pipelineListener(customPipelineListener(result.getJobId()));
  46.         Job job = jobBuilder.build();
  47.         JobReport jobReport = jobExecutor.execute(job);
  48.         result.setJobReport(jobReport.getStatus().toString());
  49.         return result;
  50.     }
  53. }
  1. @Data
  2. @AllArgsConstructor
  3. @Slf4j
  4. public class BatchResult {
  6.     private String jobId;
  7.     private long xxlJobId;
  8.     /**
  9.      * 总分发记录数
  10.      */
  11.     private Long totalNum;
  12.     /**
  13.      * 当前分发记录序号
  14.      */
  15.     private Long currentNum;
  16.     /**
  17.      * 单次写请求的数据量
  18.      */
  19.     private Integer batchSize;
  20.     private Boolean success;
  21.     private String msg;
  22.     private String jobReport;
  23.     private int successNum;
  24.     private int errorNum;
  25. }