Apache-Sparkデータフレームの距離の合計

Question

次のコードは、以下のように各列に3つの値を持つデータフレームを提供します。上記のコードの

import org.graphframes._ 
    import org.apache.spark.sql.DataFrame 
    val v = sqlContext.createDataFrame(List(
     ("1", "Al"), 
     ("2", "B"), 
     ("3", "C"), 
     ("4", "D"), 
     ("5", "E") 
    )).toDF("id", "name") 

    val e = sqlContext.createDataFrame(List(
     ("1", "3", 5), 
     ("1", "2", 8), 
     ("2", "3", 6), 
     ("2", "4", 7), 
     ("2", "1", 8), 
     ("3", "1", 5), 
     ("3", "2", 6), 
     ("4", "2", 7), 
     ("4", "5", 8), 
     ("5", "4", 8) 
    )).toDF("src", "dst", "property") 
val g = GraphFrame(v, e) 
val paths: DataFrame = g.bfs.fromExpr("id = '1'").toExpr("id = '5'").run() 
paths.show() 
val df=paths 
df.select(df.columns.filter(_.startsWith("e")).map(df(_)) : _*).show 

出力が上記出力において以下::

+-------+-------+-------+              
    |  e0|  e1|  e2| 
    +-------+-------+-------+ 
    |[1,2,8]|[2,4,7]|[4,5,8]| 
    +-------+-------+-------+ 

与えられ、我々は、各列は3つの値があり、次のように、それらが解釈できることがわかります。

e0 : 
source 1, Destination 2 and distance 8 

e1: 
source 2, Destination 4 and distance 7 

e2: 
source 4, Destination 5 and distance 8 

基本的 e0

、e1、及びe3がエッジです。各列の3番目の要素を合計したい、つまり合計距離を取得するために各エッジの距離を加算します。どうすればこれを達成できますか？

Answer 1

それは次のように行うことができます。

val total = df.columns.filter(_.startsWith("e")) 
.map(c => col(s"$c.property")) // or col(c).getItem("property") 
.reduce(_ + _) 

df.withColumn("total", total) 

Answer 2

私は列のコレクションは、合計して、UDFにfoldLeftを使用するようになるだろう：

scala> val df = Seq((Array(1,2,8),Array(2,4,7),Array(4,5,8))).toDF("e0", "e1", "e2") 
df: org.apache.spark.sql.DataFrame = [e0: array<int>, e1: array<int>, e2: array<int>] 

scala> df.show 
+---------+---------+---------+ 
|  e0|  e1|  e2| 
+---------+---------+---------+ 
|[1, 2, 8]|[2, 4, 7]|[4, 5, 8]| 
+---------+---------+---------+ 

scala> val colsToSum = df.columns 
colsToSum: Array[String] = Array(e0, e1, e2) 

scala> val accLastUDF = udf((acc: Int, col: Seq[Int]) => acc + col.last) 
accLastUDF: org.apache.spark.sql.UserDefinedFunction = UserDefinedFunction(<function2>,IntegerType,List(IntegerType, ArrayType(IntegerType,false))) 

scala> df.withColumn("dist", colsToSum.foldLeft(lit(0))((acc, colName) => accLastUDF(acc, col(colName)))).show 
+---------+---------+---------+----+ 
|  e0|  e1|  e2|dist| 
+---------+---------+---------+----+ 
|[1, 2, 8]|[2, 4, 7]|[4, 5, 8]| 23| 
+---------+---------+---------+----+