0

500 万を超える要素を繰り返し実行する Spark アプリケーションがあります。アプリケーションをデータセット全体で実行するには 2 時間かかります。しかし、5,000 万を超える要素のデータセット全体に対してアプリケーションを実行する必要があります。
コードは正常に実行されますが、問題は、私のプログラムのほとんどがドライバーで実行され、エグゼキューターがアプリケーションの実行において果たす役割は最小限です。したがって、この反復アプリケーションの計算時間は非常に長くなります。
アプリケーションは、n-triples データセットからグラフを作成して、連結要素を見つけます。
問題は、executor がタスクを受信して​​おらず、最初の for ループが 500 万個の要素すべてが完了するまで実行されるため、この部分に約 90% の時間がかかるため、主にこの部分を最適化する必要があることです。
作業をドライバーからエグゼキューターに転送する変更を提案し、それによってこのコードをスケーラブルにして、計算時間を大幅に短縮します。

import scala.io.Source 
import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext._
import org.apache.spark.graphx._
import org.apache.spark.rdd.RDD
import scala.collection.mutable.ListBuffer
import scala.collection.mutable.HashMap
import scala.collection.mutable.ArrayBuffer

object Wisdom {

val componentLists = HashMap[VertexId, ListBuffer[VertexId]]()
val prefLabelMap =  HashMap[VertexId, String]()

def main(args: Array[String]) {

val conf = new SparkConf()
val sc = new SparkContext(conf)

val tripleEndingPattern = """\s*\.\s*$""".r    
val languageTagPattern = "@[\\w-]+".r    

var edgeArray = Array(Edge(0L,0L,"http://dummy/URI"))
var literalPropsTriplesArray = new Array[(Long,Long,String)](0)
var vertexArray = new Array[(Long,String)](0)

val source = sc.textFile("hdfs://ec2-54-172-85-190.compute-1.amazonaws.com:54310/akshat/datas.nt")
val lines = source.toArray

var vertexURIMap = new HashMap[String, Long];

var triple = new Array[String](3)
var nextVertexNum = 0L
for (i <- 0 until lines.length) {

    lines(i) = tripleEndingPattern.replaceFirstIn(lines(i)," ")  
    triple = lines(i).mkString.split(">\\s+")       
    val tripleSubject = triple(0).substring(1)   
    val triplePredicate = triple(1).substring(1) 
    if (!(vertexURIMap.contains(tripleSubject))) {
        vertexURIMap(tripleSubject) = nextVertexNum
        nextVertexNum += 1
    }
    if (!(vertexURIMap.contains(triplePredicate))) {
        vertexURIMap(triplePredicate) = nextVertexNum
        nextVertexNum += 1
    }
    val subjectVertexNumber = vertexURIMap(tripleSubject)
    val predicateVertexNumber = vertexURIMap(triplePredicate)

    if (triple(2)(0) == '<') { 
        val tripleObject = triple(2).substring(1)   
        if (!(vertexURIMap.contains(tripleObject))) {
            vertexURIMap(tripleObject) = nextVertexNum
            nextVertexNum += 1
        }
        val objectVertexNumber = vertexURIMap(tripleObject)
        edgeArray = edgeArray :+
            Edge(subjectVertexNumber,objectVertexNumber,triplePredicate)
    }
    else {
        literalPropsTriplesArray = literalPropsTriplesArray :+
            (subjectVertexNumber,predicateVertexNumber,triple(2))
    }
}

for ((k, v) <- vertexURIMap) vertexArray = vertexArray :+  (v, k)  

for (i <- 0 until literalPropsTriplesArray.length) {
    if (literalPropsTriplesArray(i)._2 ==
        vertexURIMap("http://www.w3.org/2000/01/rdf-schema#label")) {

        val prefLabel =
            languageTagPattern.replaceFirstIn(literalPropsTriplesArray(i)._3,"")
        prefLabelMap(literalPropsTriplesArray(i)._1) = prefLabel;
    }
}

val vertexRDD: RDD[(Long, String)] = sc.parallelize(vertexArray)

val edgeRDD: RDD[Edge[(String)]] =
    sc.parallelize(edgeArray.slice(1,edgeArray.length))

val literalPropsTriplesRDD: RDD[(Long,Long,String)] =
    sc.parallelize(literalPropsTriplesArray)

val graph: Graph[String, String] = Graph(vertexRDD, edgeRDD)

val skosRelatedSubgraph =
    graph.subgraph(t => t.attr ==
                   "http://purl.org/dc/terms/subject")

val ccGraph = skosRelatedSubgraph.connectedComponents()

ccGraph.vertices.saveAsTextFile("hdfs://ec2-54-172-85-190.compute-1.amazonaws.com/akshat/outp")

sc.stop
}
}
4

1 に答える 1

0

RDD（コード内のソース）でマップを使用する必要があるforループを使用しています。次に、実行者が登場し、タスクが共有されます。Spark は、 for ループを使用して入力ファイルの行を反復処理するようには設計されていません。基礎をしっかり身につけてください。楽しい学習

于 2016-06-22T08:01:28.210 に答える