数据采集ETL工具 Elasticsearch-datatran v6.3.7 发布。

Elasticsearch-datatran 由 bboss 开源的数据采集同步ETL工具，提供数据采集、数据处理和数据入库功能。支持在Elasticsearch、关系数据库(mysql,oracle,db2,sqlserver、达梦等)、Mongodb、HBase、Hive、Kafka、文本文件、SFTP/FTP多种数据源之间进行海量数据采集同步；支持本地/ftp日志文件实时增量采集到kafka/elasticsearch/database；支持根据字段进行数据记录切割；支持根据文件路径信息将不同文件数据写入不同的数据库表。

提供自定义处理采集数据功能，可以按照自己的要求将采集的数据处理到目的地，支持数据来源包括：database，elasticsearch，kafka，mongodb，hbase，file，ftp等，想把采集的数据保存到什么地方，由自己实现CustomOutPut接口处理即可。

Elasticsearch版本兼容性：支持各种Elasticsearch版本（1.x,2.x,5.x,6.x,7.x,+）之间相互数据迁移

v6.3.7 变更记录

1. elasticsearch客户端改进：多数据源支持数据源引用功能，如果两个数据源都指向同一个数据源，则可以将第二个数据源指向第一个数据源，配置示例：

普通项目

elasticsearch.referExternal=default

spring boot项目

spring.elasticsearch.bboss.elasticsearch.referExternal=default

2. 数据源同步改进：增加自定义定时同步调度机制，可以指定作业执行的时间段（支持指定多个时间段）和忽略执行时间段（支持指定多个时间段），使用案例：

        //定时任务配置，
        importBuilder.setScheduleSelf()//使用bboss自带的定时器,bboss timer
                .setDeyLay(1000L) // 任务延迟执行deylay毫秒后执行
                .setPeriod(1*60*1000l)//每隔period毫秒执行，如果不设置，只执行一次
                .addScanNewFileTimeRange("12:37-23:59");//添加每天调度执行的时间段，可以调用多次addScanNewFileTimeRange方法添加多个时间段
                //添加每天排除的时间段（不调度执行作业），可以调用多次addSkipScanNewFileTimeRange方法添加多个时间段,设置addScanNewFileTimeRange，则SkipScanNewFileTimeRange不起作用
//                .addSkipScanNewFileTimeRange("11:30-13:00");
        //定时任务配置结束

如果是Filelog插件，还需要额外指定：

        FileImportConfig config = new FileImportConfig();
        /**
        * 设置是否采用外部新文件扫描调度机制：bboss timer,jdk timer,quartz,xxl-job
        * true 采用，false 不采用，默认false
        */
        config.setUseETLScheduleForScanNewFile(true);        

3. 在任务CallInterceptor.preCall中，可以根据taskContext中对应的不同的文件指定不同数据库添加、修改、删除sql,使用参考案例：

   //导出到数据源配置
        DBConfigBuilder dbConfigBuilder = new DBConfigBuilder();
        dbConfigBuilder
                .setSqlFilepath("sql-dbtran.xml")//指定sql配置文件地址
                .setTargetDbName("test");//指定目标数据库，在application.properties文件中配置
   
        importBuilder.setOutputDBConfig(dbConfigBuilder.buildDBImportConfig());
        importBuilder.addCallInterceptor(new CallInterceptor() {
            @Override
            public void preCall(TaskContext taskContext) {
                FileTaskContext fileTaskContext = (FileTaskContext)taskContext;
                String filePath = fileTaskContext.getFileInfo().getOriginFilePath();
                /**
                 * 根据文件名称指定插入数据库的sql语句
                 */
                if(filePath.endsWith("metrics-report.log")) {
                    DBConfigBuilder dbConfigBuilder = new DBConfigBuilder();
                    dbConfigBuilder.setInsertSqlName("insertSql");//指定新增的sql语句名称，在配置文件中配置：sql-dbtran.xml
   
                    taskContext.setDbmportConfig(dbConfigBuilder.buildDBImportConfig());
                }
            }
   
            @Override
            public void afterCall(TaskContext taskContext) {
   
            }
   
            @Override
            public void throwException(TaskContext taskContext, Exception e) {
   
            }
        });

采集日志文件自定义处理案例

bboss数据采集ETL案例大全

https://esdoc.bbossgroups.com/#/bboss-datasyn-demo