使用开源jdbc框架导入数据过慢

由于项目需要,我们用开源的jdbc工具代替传统loading job 方式进行数据的导入。参见了这个方法:https://github.com/tigergraph/ecosys/blob/master/etl/tg-jdbc-driver/tg-jdbc-examples/src/main/java/com/tigergraph/jdbc/UpsertQuery.java
链接中的第三种使用方式没有看懂,其中“job load_pagerank(line)” 这种job是如何编写的呢?我才用了第二种方式,即Upsert vertices and edges with PreparedStatement.数据可以正常导入,我具体是采用按行读文件,读完一行后即addBatch一次,然后10000行进行一次提交。addBatch过程比较慢,10000行数据约需3分钟。请问这正常吗?有基于jdbc导入数据更好的解决方案吗?

暂且回复您这边 可以参考下
https://github.com/tigergraph/ecosys/tree/master/etl/tg-jdbc-driver#how-to-use-in-apache-spark
性能还是不错的

你好!vertex和edge的upsert只用于adhoc的数据更新,如果需要更新大量的数量,请参考 “创建数据加载任务” 创建job。推荐使用spark进行数据加载:

val df = sc.textFile("/path/to/your_file", 100).toDF()

// invoke loading job
df.write.mode("append").format("jdbc").options(
 Map(
   "driver" -> "com.tigergraph.jdbc.Driver",
   "url" -> "jdbc:tg:http://127.0.0.1:14240",
   "username" -> "tigergraph",
   "password" -> "tigergraph",
   "graph" -> "ldbc_snb",
   "dbtable" -> "job load_ldbc_snb", // loading job name
   "filename" -> "v_comment_file", // filename defined in the loading job
   "sep" -> "|", // separator between columns
   "eol" -> "\n", // End Of Line
   "schema" -> "value", // column definition, each line only has one column
   "batchsize" -> "10000",
   "debug" -> "0")).save()

根据需要修改脚本中的参数,并保存为文件"write.scala",之后调用如下:
/path/to/spark/bin/spark-shell --jars /path/to/tg-jdbc-driver-1.2.jar -i write.scala

这种方式可以达到每小时几十GB的加载的速度。

好的,已收到邮件,谢谢。

您好,我按照您提示的方法试了,spark启动任务报错了。请问你们遇到过类似的问题吗?我个人对spark不是很熟悉,请问这个错误是jar包问题还是spark环境问题所致呢?这里的jar包我是通过mvn打包获取的。代码以及相关pom里的依赖都没有改过。

TigerGraph里有名称为“global_temp”的图吗?可以看看gsql 'ls’的输出。
方便的话,把write.scala贴出来更好。

val df = sc.textFile("test1008.csv", 100).toDF()

// invoke loading job
df.write.mode("append").format("jdbc").options(
 Map(
   "driver" -> "com.tigergraph.jdbc.Driver",
   "url" -> "jdbc:tg:http://172.21.21.100:14240",
   "username" -> "tigergraph",
   "password" -> "tigergraph",
   "graph" -> "UPTXS",
   "dbtable" -> "job loadUnionpayTrade", // loading job name
   "filename" -> "file1", // filename defined in the loading job
   "sep" -> "\t", // separator between columns
   "eol" -> "\n", // End Of Line
   "schema" -> "value", // column definition, each line only has one column
   "batchsize" -> "10000",
   "debug" -> "0")).save()

loadjob:
use graph UPTXS
begin
create loading job loadUnionpayTrade for graph UPTXS{

	define filename file1;
	
	load file1 to vertex Card values($0,$0,_,_);
	load file1 to vertex Card values($4,$4,_,$5);
	LOAD file1 TO VERTEX LinkTransfer VALUES($6, $6, $3, $1, _);
	load file1 to edge transfer values($4,$0,_,_,_,$3,_,_);
	load file1 to edge f_transfer values($4,$6);
	load file1 to edge t_transfer values($6,$0);

}
end

这一行具体是什么意思没有看懂,麻烦顺便解释一下,谢谢!

“schema"指的是源文件里每一列的含义,如下写法中DataFrame中只有一列,所以直接用”value“指代就可以了,这种效率是最好的:

val df = sc.textFile("test1008.csv", 100).toDF()

否则的话,如果df中有多列,Spark需要调用多次 PreparedStatement.setString() 来传递每一列的值,然后JDBC又需要按照顺序重新拼凑为一行,再调用loading job,效率会很差。

现在执行还有错误吗?

我就是按照这样写的,spark报错,定位不到问题所在。 可以帮忙看一下我的loading job和write.scala写的是否有问题吗?

这是由于httpclient.jar包冲突。请通过如下命令查看CLASSPATH设置:

env | grep -i classpath

可通过如下命令清除设置:

unset CLASSPATH

然后重新执行加载试一下。

你好,已解决,谢谢!速度大概5min/G,确实是jar包冲突了。
httpclient.jar是spark的默认jar,运行清除命令会有问题,我这里的解决方案是在项目中,对依赖的httpclient包做shade处理。
这里还有一个问题需请教:
基于JDBC的upsert方式和这次的spark方式加载数据,都遇到了中文字符导入图数据库后变成“??”,而传统的在gsql客户端运行 run loading job 的方式则中文正常。请问这个有好的解决方式吗?

如果源数据文件和TigerGraph位于同一台机器的不同磁盘,加载速度会快很多,我的测试结果大概是80GB/h.

请使用如下JDBC包,增加了对utf8的支持:
https://drive.google.com/open?id=1hL7j6f1_oGyON7LgGeDTt4P3VnVPr8GQ

如上述链接无法下载,请使用:
链接: https://pan.baidu.com/s/1EHuwpQfSJH02bioQMnw1Hg
提取码: 56ed

之后代码会提交至ecosys.

好的,感谢支持!

客气!如果还有问题,欢迎随时沟通!

@yong.tan

您好,由於Project需要將資料從Azure Data Lake導入TigerGraph,目前有兩種手法。

  1. 透過Databricks(Spark) + JDBC將資料加載至TigerGraph(VM) cluster
  2. 先將Azure Data Lake檔案透過Databricks(Spark) export出來轉成CSV格式之後,在將資料加載至TigerGraph(VM) cluster

請問您推薦哪一種手法呢? JDBC是否會有效能或是擴展性上的瓶頸呢?
謝謝

推荐使用方案二。