数据采集 ETL 工具 Elasticsearch-datatran v6.3.7 发布
数据采集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 变更记录
- elasticsearch客户端改进：多数据源支持数据源引用功能，如果两个数据源都指向同一个数据源，则可以将第二个数据源指向第一个数据源，配置示例：
普通项目
elasticsearch.referExternal=default
spring boot项目
spring.elasticsearch.bboss.elasticsearch.referExternal=default
- 数据源同步改进：增加自定义定时同步调度机制，可以指定作业执行的时间段（支持指定多个时间段）和忽略执行时间段（支持指定多个时间段），使用案例：
//定时任务配置，
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);
-
在任务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) { } });
