在使用datax的oraclewriter时，由于对oracle的不熟悉，以及c++编译的不熟悉，颇费了一些周折。在此，记录一下，供再次使用的人参考。
1.oracleWriter ：oracle提供了OCCI接口，便于直接往oracle里load数据，但是是c++的接口，所以，datax的oracleWriter通过对cpp代码的包装，使用JNI的方式去调用。
2.oracleJdbcWriter使用起来就简单多了，后面附上代码，不再赘述。
准备工作为：oracle客户端的安装和liboraclewriter.so的编译。
一：oracle客户端安装。本位在redhat64上使用了oracle11g，以下为详细安装步骤。
1.oracle官网下载rpm包，进行安装。
官网地址：http://www.oracle.com/technetwork/topics/linuxx86-64soft-092277.html
下载包：
oracle-instantclient11.2-basic-11.2.0.1.0-1.x86_64.rpm
oracle-instantclient11.2-devel-11.2.0.1.0-1.x86_64.rpm
oracle-instantclient11.2-sqlplus-11.2.0.1.0-1.x86_64.rpm
安装命令：
rpm -ivh oracle-instantclient11.2-basic-11.2.0.1.0-1.x86_64.rpm
rpm -ivh oracle-instantclient11.2-sqlplus-11.2.0.1.0-1.x86_64.rpm
rpm -ivh oracle-instantclient11.2-devel-11.2.0.1.0-1.x86_64.rpm
环境变量配置：
/etc/profile,追加以下内容
export ORACLE_HOME=/usr/lib/oracle/11.2/client64
export TNS_ADMIN=$ORACLE_HOME/network/admin
export NLS_LANG='AMERICAN_AMERICA.UTF8'
export LD_LIBRARY_PATH=$ORACLE_HOME/lib
export PATH=$ORACLE_HOME/bin:$PATH
保存，source /etc/profile 使之生效。
至此，oracle的安装完毕。
/**
因为对oracle的不熟悉，补充几句,以免耽误时间找各种配置。
sqlplus 使用时，网上很多资料讲要配置$TNS_ADMIN下的tnsnames.org。
其实，OCCI使用时，oracleWriter的代码里，我们使用以下方式去连接，是不需要配置的。
不需配置tnsnames.ora的形式:
java代码中： logon = username + "/" + password + "@//" + ip + ":" + port + "/" + dbname;
命令行中 ： sqlplus user/pass@//ip:port/db
格式不对时，会报以下错误：
ERROR:
ORA-12541: TNS:no listener
*/
二：liboraclewriter.so的编译及使用，以下为详细安装步骤。
1. 将datax下oracledumper下的代码，下载到linux机上，准备编译。（在之前装好oracle客户端的机器上）
2. 如果本地代码使用的包名不同于源码的，需要修改对应的文件。
修改文件：include中：xx_OracleWriterJni.h（修改文件名和 OracleWriterJni.h中方法的包路径名。）
src中：xx_OracleWriterJni.cpp (修改文件名)
3.修改src下Makefile 文件（不熟c++编译的切记）。
注意点：
根据错误提示，有一个变量应该需要追加const
-Wl,-rpath so文件运行时，依赖的包路径。确保此路径准确，oracle安装后，确认路径正确对应。
INCLUDE=-I../include -I$$JAVA_HOME/include/linux -I$$JAVA_HOME/include/
LIBS=-lclntsh -liconv -L../lib -L${ORACLE_HOME}/lib -L../../../../libs/ -L/usr/lib/oracle/11.2/client64/lib/
CC=g++
OBJS=liboraclewriter.so
CFLAGS=-shared -fPIC -Wl,-rpath=/usr/lib/oracle/11.2/client64/lib/
CPP=common.cpp dumper.cpp oradumper.cpp strsplit.cpp com_suning_dc_cybertron_datax_plugins_writer_oraclewriter_OracleWriterJni.cpp OBJS: $(CPP)
$(CC) $(INCLUDE) -o $(OBJS) $(CPP) $(CFLAGS) $(LIBS)
clean:
rm -rf $(OBJS)
4.make 生成liboraclewriter.so。（make环境缺少g++等，自行安装，不赘述）。
至此，可以happy的去代码中使用适合本地环境的occi接口的oraclewriter了，其他几个so包使用源码中自带的就好。
System.load(PropertyReader.getJETFIRE_HOME() + "libs"
+ "/libcommon.so");
System.load(PropertyReader.getJETFIRE_HOME() + "libs"
+ "/libcharset.so");
System.load(PropertyReader.getJETFIRE_HOME() + "libs"
+ "/libiconv.so.2");
System.load(PropertyReader.getJETFIRE_HOME() + "libs"
+ "/liboraclewriter.so");
OCCI 和 jdbc writer的性能对比：
occi :
Total time costs : 142s
Average byte speed : 1MB/s
Average line speed : 35211L/s
Total transferred records : 5000000
Total discarded records : 0
jdbc:
Total time costs : 2229s
Average byte speed : 68KB/s
Average line speed : 2243L/s
Total transferred records : 5000000
问题及异常：
.OCI使用的是direct-insert的方式，需要锁表，有时报错如下： 现象：
OCI Error - occurred at File oradumper.cpp:
Error[] - ORA-: error occurred at recursive SQL level
ORA-: insufficient privileges status: -, , oradumper.cpp, init_load
, oradumper.cpp, init_load 对策：
oracle上除了赋予用户insert select权限以外，还要
grant lock any table to 你的用户。
附:jdbcWriter代码
public class OracleJdbcWriter extends Writer {
private Logger logger = Logger.getLogger(OracleJdbcWriter.class);
private String password;
private String username;
private String dbname;
private String table;
private String pre;
private String post;
private String encoding;
private int limit;
private int failCount;// count error lines
private long concurrency;
private int batchSize;
private String sourceUniqKey = "";
private String port;
private String insert;
private String host;
private String DRIVER_NAME = "oracle.jdbc.driver.OracleDriver";
private Connection connection;
private String writeColumns;
@Override
public int init() {
password = param.getValue(ParamKey.password, "");
username = param.getValue(ParamKey.username, "");
host = param.getValue(ParamKey.ip);
port = param.getValue(ParamKey.port, "3306");
dbname = param.getValue(ParamKey.dbname, "");
table = param.getValue(ParamKey.table, "");
pre = param.getValue(ParamKey.pre, "");
post = param.getValue(ParamKey.post, "");
insert = param.getValue(ParamKey.insert, "");
encoding = param.getValue(ParamKey.encoding, "UTF-8");
limit = param.getIntValue(ParamKey.limit, 1000);
concurrency = param.getIntValue(ParamKey.concurrency, 1);
writeColumns = param.getValue(ParamKey.writeColumns, "");
batchSize = param.getIntValue(ParamKey.batchSize, 50000);
this.sourceUniqKey = DBSource.genKey(this.getClass(), host, port,
dbname);
this.host = param.getValue(ParamKey.ip);
this.port = param.getValue(ParamKey.port, "3306");
this.dbname = param.getValue(ParamKey.dbname);
return PluginStatus.SUCCESS.value();
}
@Override
public int prepare(PluginParam param) {
this.init();
this.setParam(param);
DBSource.register(this.sourceUniqKey, this.genProperties());
if (StringUtils.isBlank(this.pre))
return PluginStatus.SUCCESS.value();
Statement stmt = null;
try {
this.connection = DBSource.getConnection(this.sourceUniqKey);
stmt = this.connection.createStatement(
ResultSet.TYPE_SCROLL_INSENSITIVE,
ResultSet.CONCUR_UPDATABLE);
for (String subSql : this.pre.split(";")) {
this.logger.info(String.format("Excute prepare sql %s .",
subSql));
stmt.execute(subSql);
}
this.connection.commit();
return PluginStatus.SUCCESS.value();
} catch (Exception e) {
throw new DataExchangeException(e.getCause());
} finally {
try {
if (null != stmt) {
stmt.close();
}
if (null != this.connection) {
this.connection.close();
this.connection = null;
}
} catch (SQLException e) {
}
}
}
@Override
public int connect() {
return PluginStatus.SUCCESS.value();
}
@Override
public int startWrite(LineReceiver receiver) {
PreparedStatement ps = null;
try {
this.connection = DBSource.getConnection(this.sourceUniqKey);
Line line = null;
int lines = 0;
line = receiver.getFromReader();
if (line == null) {// 读取数据为空的情况。
return PluginStatus.SUCCESS.value();
}
this.insert = buildInsertStr(line);
ps = this.connection.prepareStatement(this.insert,
ResultSet.TYPE_SCROLL_INSENSITIVE,
ResultSet.CONCUR_UPDATABLE);
this.connection.setAutoCommit(false);
while (line != null) {
try {
for (int i = 0; i < line.getFieldNum(); i++) {
ps.setObject(i + 1, line.getField(i));
}
ps.execute();
} catch (SQLException e) {
logger.error("Invalid Data: " + line.toString(','));
logger.error(e.getMessage());
failCount++;
if (failCount >= this.limit) {
logger.error("出错条数 (" + failCount + ") 超过限制条数。");
e.printStackTrace();
throw new DataExchangeException(e);
} else {
continue;
}
}
if (lines++ == this.batchSize) {
logger.info(lines + " committed by worker "
+ Thread.currentThread().getName() + " .");
lines = 0;
this.connection.commit();
}
line = receiver.getFromReader();
}
this.connection.commit();
this.connection.setAutoCommit(true);
this.getMonitor().setFailedLines(this.failCount);
return PluginStatus.SUCCESS.value();
} catch (Exception e2) {
throw new DataExchangeException(e2.getCause());
} finally {
if (null != ps)
try {
ps.close();
} catch (SQLException e3) {
}
}
}
@Override
public int post(PluginParam param) {
if (StringUtils.isBlank(this.post))
return PluginStatus.SUCCESS.value();
Statement stmt = null;
try {
this.connection = DBSource.getConnection(this.sourceUniqKey);
stmt = this.connection.createStatement(
ResultSet.TYPE_SCROLL_INSENSITIVE,
ResultSet.CONCUR_UPDATABLE);
for (String subSql : this.post.split(";")) {
this.logger.info(String.format("Excute prepare sql %s .",
subSql));
stmt.execute(subSql);
}
return PluginStatus.SUCCESS.value();
} catch (Exception e) {
e.printStackTrace();
throw new DataExchangeException(e.getCause());
} finally {
try {
if (null != stmt) {
stmt.close();
}
if (null != this.connection) {
this.connection.close();
this.connection = null;
}
} catch (Exception e2) {
}
}
}
@Override
public List<PluginParam> split(PluginParam param) {
OracleJdbcWriterSplitter splitter = new OracleJdbcWriterSplitter();
splitter.setParam(param);
splitter.init();
return splitter.split();
}
@Override
public int commit() {
return PluginStatus.SUCCESS.value();
}
@Override
public int finish() {
return PluginStatus.SUCCESS.value();
}
private Properties genProperties() {
Properties p = new Properties();
p.setProperty("driverClassName", this.DRIVER_NAME);
String url = "jdbc:oracle:thin:@" + this.host + ":" + this.port + "/"
+ this.dbname;
p.setProperty("url", url);
p.setProperty("username", this.username);
p.setProperty("password", this.password);
p.setProperty("maxActive", String.valueOf(this.concurrency + 2));
return p;
}
private String buildInsertStr(Line line) {
String sql = "";
String[] subcos = null;
// 判断到底是全部插入表还是部分插入,最终确定sql
if (!writeColumns.equals("")) {
// 如果是部分插入的话
sql = "insert into " + table + "(";
subcos = writeColumns.split(",");
sql += writeColumns;
sql += ") values(";
for (int i = 0; i < subcos.length; i++) {
sql += "?";
sql += ",";
}
sql = sql.substring(0, sql.length() - 1);
sql += ")";
} else {
sql = "insert into " + table + " values(";
for (int i = 0; i < line.getFieldNum(); i++) {
sql += "?";
sql += ",";
}
sql = sql.substring(0, sql.length() - 1);
sql += ")";
}
return sql;
}
}