gfwleak/wanglihui-ip-learning-graph
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

自定义ArangoRDD

Browse Source
This commit is contained in:
wanglihui
2020-10-26 09:54:14 +08:00
parent c211d99c2e
commit db5ca9db08
2 changed files with 6 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
ip-learning-spark/src/main/scala/cn/ac/iie/spark/rdd/ArangoRdd.scala
View File

@@ -46,7 +46,7 @@ class ArangoRdd[T: ClassTag](@transient override val sparkContext: SparkContext,
LOG.info(sql) LOG.info(sql)
arangoCursor = arangoDB.db(options.database).query(sql,clazz.runtimeClass.asInstanceOf[Class[T]]) arangoCursor = arangoDB.db(options.database).query(sql,clazz.runtimeClass.asInstanceOf[Class[T]])
}catch { }catch {
case e: Exception => LOG.error("创建Cursor异常") case e: Exception => LOG.error(s"创建Cursor异常:${e.getMessage}")
}finally { }finally {
arangoDB.shutdown() arangoDB.shutdown()
} }
@@ -66,13 +66,13 @@ class ArangoRdd[T: ClassTag](@transient override val sparkContext: SparkContext,
private def getCountTotal: Long = { private def getCountTotal: Long = {
val arangoDB = spark.createArangoBuilder(options).build() val arangoDB = spark.createArangoBuilder(options).build()
var cnt = 0L var cnt = 0L
val sql = "RETURN LENGTH(" + options.collection + ")" val sql = s"RETURN LENGTH(${options.collection})"
LOG.info(sql) LOG.info(sql)
try { try {
val longs = arangoDB.db(options.database).query(sql, classOf[Long]) val longs = arangoDB.db(options.database).query(sql, classOf[Long])
while (longs.hasNext) cnt = longs.next while (longs.hasNext) cnt = longs.next
} catch { } catch {
case e: Exception => LOG.error(sql + "执行异常") case e: Exception => LOG.error(sql + s"执行异常:${e.getMessage}")
}finally { }finally {
arangoDB.shutdown() arangoDB.shutdown()
} }

3
ip-learning-spark/src/test/scala/cn/ac/iie/spark/RDDTest.scala
View File

@@ -27,6 +27,9 @@ object RDDTest {
doc.addAttribute("abc", 1) doc.addAttribute("abc", 1)
doc doc
}) })
value.map(doc => {(doc.getKey,doc)})
value.persist(StorageLevel.MEMORY_AND_DISK) value.persist(StorageLevel.MEMORY_AND_DISK)
value.foreach(row => println(row.toString)) value.foreach(row => println(row.toString))