datax+gbload+GBase入库适配操作手顺

一、环境准备

1、开启vsftpd服务，注意修改，默认路径（gbload_server_9.5.25版本支持sftp协议）

userlist_enable=YES -- 设定userlist 可用，一般是deny。这个不强求。如果是，将对应用户从列表注销(ftpuser、user_list)

local_root=/opt/gbload -- 设定一个默认目录。对应下面 gbload_ftp_server_base_path 这个路径。

chroot_local_user=YES -- 限制在一个目录。

allow_writeable_chroot=YES -- 这个目录可写

2、安装gccli客户端

注意修改环境变量，参考配置文件

3、dispcli （区分python2 和3 需要申请特定注意。要python某个版本编译的）

放入目录，比如 /opt/dispcli （这里里面才是dispcli文件）

4、Java环境变量 （注意将sgloader 和jdbc 也配置进去。不一定有用，配置了肯定可以）

export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.131-11.b12.el7.x86_64

export PATH=.:$PATH:$JAVA_HOME/bin

export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar:/opt/gbloader_test/ gbase-connector-java-8.3.81.53-build55.4.1-bin.jar:/opt/gbloader_test/sgLoader/sgloader-0.0.1.jar

二、gbloader部署

1、解压后有2个目录，api_file 、server_file ，比如放在/opt/gbload_server

2、创建目录用于放加载临时文件

mkdir -p /opt/ gbload_server /data/main

mkdir -p /opt/ gbload_server /data/copy

mkdir -p /opt/ gbload_server /data/kvstore

3、修改gbload_server的配置文件，在/opt/gbload_server/server_file/下面，名字叫gbload_server.conf

修改的几个内容如下：

gbload_config_server_host=10.10.3.233  #就是部署本机的IP地址

gbload_config_extern_master_server=10.10.3.233 #就是部署本机的IP地址，以后如果加载机有扩展的情况下，这里写的是master 的IP地址，只有一台加载机则就是本机的IP地址

gbload_data_key_retention_time=2 #表加载临时信息的保存时间，单位位秒，设置2就行；

gbload_ftp_server_host_name=10.10.3.233 #就是部署本机的IP地址

gbload_ftp_server_user_passwd=gbase:gbase  #连接本机ftp服务的用户名和密码，格式为user:passwd

gbload_ftp_server_base_path=/opt/gbloader_test/server_file/  #本机ftp服务的主目录，用户由gbload_ftp_server_user_passwd指定，加载服务必须对此目录有读写权限

gbload_dispatch_server_ip_port=10.10.3.235:8080 #dispatch服务的IP和端口

gbload_dispatch_client_path=/opt/gbloader_test/dispatch_server/ #dispcli所在的路径

gbload_gnode_gbloader_path=/opt/gbloader_test/server_file/ #gbloader所在的路径

gbload_database_server_host=10.10.3.109 #数据库节点的IP地址；需要制定本地gcluster/gnode使用的IP地址（GBase MPP Cluster的一个管理节点的IP）

gbload_file_count_per_controy_file=100000 #每个控制文件处理的数据文件个数

4、启动加载服务，在/opt/gbload_server/server_file下执行nohup sh ./ gbload_server_monit.sh &即可，并通过ps –ef|grep gbload_server看进程是否启动

5、gbload加载测试样例

1. 表结构

Create database testdb;

Use testdb;

CREATE TABLE "lineorder" (

  "lo_orderkey" bigint(20) DEFAULT NULL,

  "lo_linenumber" int(11) DEFAULT NULL,

  "lo_custkey" int(11) DEFAULT NULL,

  "lo_partkey" int(11) DEFAULT NULL,

  "lo_suppkey" int(11) DEFAULT NULL,

  "lo_orderdate" int(11) DEFAULT NULL,

  "lo_orderpriority" varchar(15) DEFAULT NULL,

  "lo_shippriority" varchar(1) DEFAULT NULL,

  "lo_quantity" int(11) DEFAULT NULL,

  "lo_extendedprice" int(11) DEFAULT NULL,

  "lo_ordtotalprice" int(11) DEFAULT NULL,

  "lo_discount" int(11) DEFAULT NULL,

  "lo_revenue" int(11) DEFAULT NULL,

  "lo_supplycost" int(11) DEFAULT NULL,

  "lo_tax" int(11) DEFAULT NULL,

  "lo_commitdate" int(11) DEFAULT NULL,

  "lo_shipmode" varchar(10) DEFAULT NULL

) COMPRESS(5, 5) ENGINE=EXPRESS DISTRIBUTED BY('lo_orderkey');

1. Java样例代码

以下代码用的时候，修改红色加粗的IP地址就行，就是运行gbload_server的这台机器的IP地址。

url = "jdbc:gbload://" + host + ":" + port + "/" + db  + "?failoverEnable=true&hostList=" + "10.10.3.233" + "&basePath=" + sgLoader + "&sgloaderPath=" + sgLoader + "&interval=120&characterEncoding=gbk&UseOneServer=false";  // URL

import java.io.*;

import java.sql.Connection;

import java.sql.DriverManager;

import java.sql.PreparedStatement;

import java.sql.SQLException;

import java.util.ArrayList;

public class Lineorder implements Runnable{

 String host; // 加载服务IP地址

 String port = "8585"; // 端口

 String port2 = "5258";

 String db;  // 数据库名

 String table;   // 要加载的表名

 int psize;   // 每次提交的行数

 int thdCount; //线程数

 static int sleepTime = 0;  //每次提交sleep的毫秒数

 String sgLoader;

 String fileName ;

 public Lineorder(String host, String db, String table, int psize, String sgLoader, String fileName){

          this.host = host;

          this.db = db;

          this.table = table;

          this.psize = psize;

          this.sgLoader = sgLoader;

          this.fileName = fileName;

 }

 @Override

 public void run() {

          // TODO Auto-generated method stub

          String url = "jdbc:gbload://" + host + ":" + port + "/" + db  + "?failoverEnable=true&hostList=" + "10.10.3.233;10.10.3.234" + "&basePath=" + sgLoader + "&sgloaderPath=" + sgLoader + "&interval=120&characterEncoding=gbk&UseOneServer=false";  // URL

          try {

           Class.forName("cn.gbase.jdbc.LoadDriver").newInstance();

          } catch (Exception e) {

                  e.printStackTrace();

                  return;

          }

          Connection conn = null;

          PreparedStatement pstmt = null;

          try {

                  System.out.println(url);

                  conn =  DriverManager.getConnection(url, "gbase", "gbase20110531"); 

                  System.out.println("getConnection done");

          } catch (SQLException e1) {

                  e1.printStackTrace();

                  return;

          }

          if(conn == null){

                  System.err.println("conn is null");

                  return;

          }

          String sql = "insert into " + table + " values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)";

          //String sql = "insert into lineorder (lo_orderkey,lo_linenumber,lo_custkey,lo_partkey,lo_suppkey,lo_orderdate,lo_orderpriority,lo_shippriority,lo_quantity,lo_extendedprice,lo_ordtotalprice,lo_discount,lo_revenue,lo_supplycost,lo_tax,lo_commitdate,lo_shipmode) values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)";

          File file = new File(fileName);

        BufferedReader reader = null;

          try{

            reader = new BufferedReader(new FileReader(file));

            String tempString = null;

            String[] lines = null;

                  pstmt = conn.prepareStatement(sql);     

                  String threadName = Thread.currentThread().getName();

                  int line = 0;

            while ((tempString = reader.readLine()) != null) {

                lines = tempString.split("\\|");

                pstmt.setString(1, lines[0]);

                pstmt.setString(2, lines[1]);

                pstmt.setString(3, lines[2]);

                pstmt.setString(4, lines[3]);

                pstmt.setString(5, lines[4]);

                pstmt.setString(6, lines[5]);

                pstmt.setString(7, lines[6]);

                pstmt.setString(8, lines[7]);

                pstmt.setString(9, lines[8]);

                pstmt.setString(10, lines[9]);

                pstmt.setString(11, lines[10]);

                pstmt.setString(12, lines[11]);

                pstmt.setString(13, lines[12]);

                pstmt.setString(14, lines[13]);

                pstmt.setString(15, lines[14]);

                pstmt.setString(16, lines[15]);

                pstmt.setString(17, lines[16]);

                pstmt.addBatch();

                line++;

                if(line == psize){

                 line = 0;

                 pstmt.executeBatch();

                 conn.commit();

                 pstmt.clearBatch(); 

                }

            }

         pstmt.executeBatch();

         conn.commit();

         pstmt.clearBatch();

            reader.close();

           System.out.println(threadName+": ==================>LOAD END! " );

          }catch(Exception e){

                  e.printStackTrace();

          }finally{

            if (reader != null) {

                try {

                    reader.close();

                } catch (IOException e) {

                 e.printStackTrace();

                }

            }

                  try {

                          if (pstmt != null)

                                   pstmt.close();

                  } catch (SQLException e) {

                          e.printStackTrace();

                  }

                  try {

                          if(conn != null)

                                   conn.close();

                  } catch (SQLException e) {

                          e.printStackTrace();

                  }

          }       

 }

 public static void main(String[] args) {

          String host = args[0]; // 加载服务IP地址

          String db = args[1];   // 数据库名

          String table=args[2];    // 要加载的表名

          int psize = Integer.parseInt(args[3]) ;

          int thdCount = Integer.parseInt(args[4]);

          String sgLoader = args[5]; //sgLoader路径

          String fileName = args[6];  //加载的文件名

          ArrayList<Thread> loadThreads = new ArrayList<Thread>(thdCount);

          long beginTime = System.currentTimeMillis();

          // 启动加载线程。

          for(int i=1; i<=thdCount; i++){

                  Lineorder gbl = new Lineorder(host,db,table,psize,sgLoader,fileName);

                  Thread thd = new Thread(gbl);

                  loadThreads.add(thd);

          }

          for (Thread t : loadThreads) {

                  t.start();

          }

          for (Thread t : loadThreads) {

                  try {

                          t.join();

                  } catch (InterruptedException e) {

                          e.printStackTrace();

                  }

          }

          long endTime = System.currentTimeMillis();

          double useTime = (endTime - beginTime) / 1000; // 秒

          //double speed = (psize * pcount * thdCount ) / useTime;

          //System.out.println("rowCount: "+ psize * pcount * thdCount );

          System.out.println("time: "+ useTime + " sec");

          //System.out.println("speed: " + speed + " row/sec");

 }

}

1. Java代码编译及执行

编译：javac Lineorder.java

执行：java Lineorder 192.168.2.188 testdb lineorder 10000 1 /opt/gbload_server/api_file/ /opt/gbload_server/data/lineorder.tbl 

执行参数说明：

Lineorder为java编译出来的class文件

192.168.2.188为gbload_server运行所在机器的IP地址；

testdb：gbase 8a MPP cluster数据库中的数据库名称；

Lineorder：表名称；

10000：每次提交的数据量，建议10000~20000，但不是绝对，根据服务器的配置灵活调整，多测试几组数据找出最有的批量提交值；

１：线程数

/opt/ gbload_server /api_file/：sgloader-0.0.1.jar所在的路径

/opt/gbload_server/data/lineorder.tbl：需要加载的数据文件

1. 测试加载，如下图所未，gbload加载成功

三、datax部署

1、解压datax.zip到／opt目录

2、解压公司的datax_plugin_build3.0.zip解压包，对应的plugin下面有reader和writer，放到datax的plugin下对应目录下面

3、用gbload_server的gcluster-load-api-8.5.1.2-build_20241217080553.jar包，datax的plugin里gbase8amppgbloaderwriter和gbase8amppjdbcwriter对应的jar包，保持版本一致（gbload_server高版本与datax自带jar包兼容有问题）

4、 python datax.py -r mysqlreader -w gbase8amppgbloaderwriter 可以生成样例，可以根据生成的样例进行修改，生成json任务文件

5、执行python datax.py ${name}.json

任务执行最后，统计信息