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加载测试样例

表结构

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');

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 loadThreads = new ArrayList(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");

}

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：需要加载的数据文件

测试加载，如下图所未，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

任务执行最后，统计信息

相关推荐

C# 13 和 .NET 9 全知道 :13 使用 ASP.NET Core 构建网站 (1)

程序员的开源月刊《HelloGitHub》第 71 期

详细介绍一下Redis的Watch机制，可以利用Watch机制来做什么?

假如有100W个用户抢一张票，除了负载均衡办法，怎么支持高并发?

如何将AI助手接入微信（打开ai手机助手）

Java面试必考问题:什么是乐观锁与悲观锁

SparkSQL——DataFrame的创建与使用

redission YYDS spring boot redission 使用

一文带你了解Redis与Memcached? redis与memcached的区别

如何利用Redis进行事务处理呢? 如何利用redis进行事务处理呢英文