码迷,mamicode.com
首页 > 数据库 > 详细

datax中oracleWriter

时间:2015-12-31 12:43:26      阅读:263      评论:0      收藏:0      [点我收藏+]

标签:

在使用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++编译的切记)。

  注意点:

-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

  

附: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;
    }

}

 

datax中oracleWriter

标签:

原文地址:http://www.cnblogs.com/drawwindows/p/5091227.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!