我用的是spark-2.12.3, nebula-2.5.0，使用docker-compose安装的，nebula-spark-connector也是从 github上pull下来的最新的。
我首先编译connector:
cd nebula-spark-utils/nebula-spark-connector
mvn clean package -Dmaven.test.skip=true -Dgpg.skip -Dmaven.javadoc.skip=true
然后把这个生成的jar文件复制到我的project/lib目录里面:
cp -riv target/*jar /path/to/project/lib
我的build.sbt是这样的:
name := "Example"
version := "0.0"
scalaVersion := "2.12.3"
libraryDependencies += "org.apache.spark" %% "spark-sql" % "3.1.2"
libraryDependencies += "org.apache.spark" %% "spark-core" % "3.1.2"
libraryDependencies += "org.apache.spark" %% "spark-mllib" % "3.1.2"
我的代码这样的 ：
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.graphx._
import org.apache.spark.rdd.RDD
import org.apache.spark
import scala.io.Source
import scala.util.control.Breaks._
import org.apache.spark.sql.SparkSession
import java.nio.charset.StandardCharsets
import collection.mutable.ArrayBuffer
import com.vesoft.nebula.connector.connector.{NebulaDataFrameReader, NebulaDataFrameWriter}
import com.vesoft.nebula.connector.{NebulaConnectionConfig, ReadNebulaConfig, WriteNebulaConfig, WriteNebulaVertexConfig}
object Example {
def main(args: Array[String]) {
try_database()
}
def try_database() {
val spark_conf = new SparkConf
spark_conf
.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
val spark = SparkSession
.builder()
.appName("test")
.master("local")
.config(spark_conf)
.getOrCreate()
val v_df = spark.read.json("../data/vertices.json")
v_df.show()
val e_df = spark.read.json("../data/edges.json")
e_df.show()
// 写入数据库的方法
val nbl_config = NebulaConnectionConfig.builder()
.withMetaAddress("10.128.61.8:49221") // docker-compose生成的容器的使用的本地端口
.withGraphAddress("10.128.61.8:49238")
.build()
val nbl_write_vert_conf = WriteNebulaVertexConfig.builder()
.withSpace("test") // db里面的space
.withTag("vert") // vertex类型
.withVidField("vertexId") // 这个df里面的表示vid的列
// .withVidPolicy("hash") // vid如果不是整数就加这个，是就不用管
.withVidAsProp(false)
.withUser("root")
.withPasswd("nebula")
.withBatch(1000)
.build()
v_df.write.nebula(nbl_config, nbl_write_vert_conf).writeVertices()
}
}
运行的时候会有一个报错:
Exception in thread "main" java.lang.NoClassDefFoundError: com/vesoft/nebula/connector/WriteNebulaConfig
at Example.main(main.scala)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.spark.deploy.JavaMainApplication.start(SparkApplication.scala:52)
at org.apache.spark.deploy.SparkSubmit.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:951)
at org.apache.spark.deploy.SparkSubmit.doRunMain$1(SparkSubmit.scala:180)
at org.apache.spark.deploy.SparkSubmit.submit(SparkSubmit.scala:203)
at org.apache.spark.deploy.SparkSubmit.doSubmit(SparkSubmit.scala:90)
at org.apache.spark.deploy.SparkSubmit$$anon$2.doSubmit(SparkSubmit.scala:1039)
at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:1048)
at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
Caused by: java.lang.ClassNotFoundException: com.vesoft.nebula.connector.WriteNebulaConfig
at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
at java.lang.ClassLoader.loadClass(ClassLoader.java:418)
at java.lang.ClassLoader.loadClass(ClassLoader.java:351)
... 13 more
请问我是哪里没弄对吗
我的理解是最后 coincheung 用的是 2.12 版本。这个帖子开了过 1 个月了，先关闭了，你有问题可以新开帖子哈~