Oracle中的大数据导入,帮这边做一个数据导入的接口,将各个文件里面的数据定时导入的Oracle表中,有如下几点要求:
1 时效性,数据导入效率不能太低。
2 多文件,这种文件接口的数量很多,估计有20多个,以后也有可能增加,而且有可能多个文件对应于同一张表,即将多个文件里面的数据导入到同一个表中。
3执行时间间隔不一定,有点文件一个月导入一次,有的则需要每天导入一次,而且有的表需要全量覆盖之前的数据,有的则需要追加到原表中。
这个接口其实并不复杂,需要做的事情,不过是读取文件,然后分析文件,讲文件提交入Oracle数据库即可 。
但是在开始之前,需要考虑以下几个事情:
3,如果出现异常如何处理。
4,如何保证高效性。
需要说明的是,从文件中逐行导入数据,有一个先决条件,即文件中每行数据的每一个字段,必须和表一一对应,有两种方法实现这个功能:
1 是在文件的第一行加一个说明行,说明行里面包含文件每行每个字段对应的列名。
2 每行的数据预先约定好,我们将其放在配置表或配置文件中,当需要解析某一个文件的时候,读取配置表,得到相应的列说明。
这次的文件接口中对端系统使用的方式是第第二种,我们这边也采取第二种方式来实现。
在问题出现之前,并不需要考虑如何解决问题,但是需要记录那些问题有可能出现 ,所以,对于问题3,我们需要记录日志,即将每一步操作记录下来。
问题4才是我们需要马上关心的问题,怎么办呢?其实也很简单, 多线程,好了,这个接口以及有合适的方案来解决了,首先,我们先完善一下我们的程序框架:
首先,我们需要一个TimerTask来作为我们的“工人”来执行我们的任务:
public class Task extends TimerTask {
@Override
public void run() {
// TODO Auto-generated method stub
}
}
现在,我们需要一个定时器
package com.ztesoft.bsn.gslocal.interfaces.tpsshb.bll;
import java.util.Calendar;
import java.util.Date;
import java.util.Timer;
public class TimerManager {
public TimerManager() {
Timer timer = new Timer();
Task task = new Task();
java.util.Date date=new java.util.Date();
Calendar calendar = Calendar.getInstance();
calendar.set(Calendar.HOUR_OF_DAY, 8);
calendar.set(Calendar.MINUTE, 32);
calendar.set(Calendar.SECOND, 0);
Calendar ca=Calendar.getInstance();
int day=ca.get(Calendar.DAY_OF_MONTH);
System.out.println(ca.get(Calendar.DAY_OF_MONTH));
//timer.schedule(task, date, 24*3600*100);
Date time = calendar.getTime();
timer.schedule(task, time, 1*300*1000);
}
public static void main(String[] args){
TimerManager tm=new TimerManager();
}
}
我们使用 Calendar 来设置定时执行的时间,每天定时执行的时间,当然,我们也可以将定时执行的时间点写入配置表中,这样我们就不要修改代码来控制程序每天执行的时间了。
现在,我们来分析一下如何实现我们的业务逻辑:
首先,我们需要将文件读入内存,再将内存中的数据写入数据库中,我们要边读边写吗?
是,也不是
考虑一下读取文件的开销 rime1 和将数据写入数据库的开销 time2, 我们会发现 time1>>time2 当然现在数据库操作也不会很慢,但是仍然大于在直接读取文件的时间开销,其实这个指标很重要,会直接影响我们的具体实现逻辑,现在,我们就以 time1 >>time2 这种情况来处理,后面也证明这样的假定是正确的,我们读取文件,当每读取 1000行,或者100行,将数据写入数据库中。
在这里,我们使用 java.sql.PreparedStatement 来完成这个操作,当我们每读取、分析一行记录之后,我们的可以用
PreparedStatement .addBatch();
这个方法来将其添加到我们的“池子里面”,然后当池子的数量达到我们的临界值 ,比如1000之后,我们将其提交,具体的代码可以是这样的:
if(readCount%process_count==0){
PreparedStatement .executeBatch();
PreparedStatement .clearBatch();
connection.commit();
}
我们假定我们的文件有1111211行记录,我们每1000行提交一次,只用上面代码是,不够的,最后会有211行记录丢失,所以,我们需要在整个循环结束之后,再执行以下如下代码:
PreparedStatement .executeBatch();
PreparedStatement .clearBatch();
connection.commit();
需要注意的是,我们使用PueparedStatement 对象来进行数据库操作,在我们的SQL语句中,不应包含有要插入的数据,这句话是什么意思呢 ?举个例子
我们要插入一条学生记录,我们的是SQL应该是
Insert into student(name,id,sex)values(?,?,?)
而不是:
Insert into student(name,id,sex)values('张三','1000010','男')
这样的方式,否则,你会发现,你只会提交“池子”里最后一条记录,感兴趣的同学可以试一下,而且我想大家也能理解其中的缘由。
接下来,我们需要考虑如何让这个接口“可配置”,也就是说,当你新增需要导入的数据是,是不用再修改代码的,如何实现呢?我们可以考虑使用两张配置表:
create table DATA_CONFIG
(
file_path VARCHAR2(400),
offen_value NUMBER,
extend_id NUMBER,
needoverwrite CHAR(1),
table_name VARCHAR2(400),
Syste_Code VARCHAR2(400)
)
create table DATA_TABLES
(
extend_id NUMBER,
table_name VARCHAR2(40),
filed_name VARCHAR2(40),
seq NUMBER,
delte_state VARCHAR2(1) default '0'
)
这两张表来配合使用,我们就可以完全我们的全部功能了,
第一张表作为我们的“接口信息表”,这张表说明了我们的接口编码(系统之间交互,通过接口编码来识别,可以实现接口的通用性)、文件路径(这个路径不必,也不应该写成绝对路径或相对路径,这应该是应“逻辑路径”,比如 ,我们的路径可以是这样的:d: emp*data_time*#(sys_Code)#.dat)
这个相对路径表示我们需要解析的文件位于:d: emp这个目录下,中间还包含一个由日期命名的子目录,最后是一个以.dat结尾的,用接口编码命名的文件。当我们处理完成一个文件之后,可以加一个我们喜欢的,或者惯例要求的后缀,比如 ".bak",每次导入数据是否要覆盖原有数据 OverWriter属性。
第二张表则是我们保存表字段的表,这个表是一张“纵表”,我们可以建立一个实体类与之对应:
public class Field {
private String Name;
public String getName() {
return Name;
}
public void setName(String name) {
Name = name;
}
public String getType() {
return Type;
}
public void setType(String type) {
Type = type;
}
public int getSeq() {
return seq;
}
public void setSeq(int seq) {
this.seq = seq;
}
private String Type;
private int seq;
private String defaultValue;
public String getDefaultValue() {
return defaultValue;
}
public void setDefaultValue(String defaultValue) {
this.defaultValue = defaultValue;
}
}
Field 类表示每一个表的每一类,有列名、列类型,列序号。public class ExtendObject {
private String file_Path;
private int offenValue;
private int extend_id;
private String table_name;
private String needovervrite;
private java.util.ArrayList<Field> Fileds;
public java.util.ArrayList<Field> getFileds() {
return Fileds;
}
public void setFileds(java.util.ArrayList<Field> fileds) {
Fileds = fileds;
}
。。。。添加Get和SET方法
}
这个实体类。
下面来完成我们的整体框架:
定时器每天定时执行,扫描我们共有多少需要同步的接口,然后每一个接口启动一个线程来进行数据同步。即,我们首先从数据库中得到一个 ExtendObject 对象的列表,然后再去执行这些对象:
for (ExtendObject obj : etlist) {
//这里为每个对象启动一个线程。
}
好了,这个接口的核心部分已经做完了,别急,我们我们还没有搞完,还有很关键的一部分——异常
首先,我们先分析一下出现异常的可能:
1 ,数据异常:透传的数据不符合要求或规范。
2,数据库操作异常:在进行数据库操作时,出现超时等错误。
要处理这一的异常,我们需要做的是:
1 ,进行日志记录,对每一个文件,异常信息应该记录的是 :出现问题的行数,问题的详细信息
2,若某个文件全部处理正常,则记录处理成功,并且将文件名做修改:如添加 ".bak"后缀,这样,第二次读到这个文件时,不会再去处理这个文件了,好了OK。