Re: parallelize for a large Seq is extreamly slow.

[Earthson]

parallelize is still so slow. 

~~~~~~~~

package com.semi.nlp

import org.apache.spark._
import SparkContext._
import scala.io.Source
import com.esotericsoftware.kryo.Kryo
import org.apache.spark.serializer.KryoRegistrator

class MyRegistrator extends KryoRegistrator {
    override def registerClasses(kryo: Kryo) {
        kryo.register(classOf[Map[String,Int]])
        kryo.register(classOf[Map[String,Long]])
        kryo.register(classOf[Seq[(String,Long)]])
        kryo.register(classOf[Seq[(String,Int)]])
    }
}

object WFilter2 {
    def initspark(name:String) = {
        val conf = new SparkConf()
                    .setMaster("yarn-standalone")
                    .setAppName(name)
                    .setSparkHome(System.getenv("SPARK_HOME"))
                    .setJars(SparkContext.jarOfClass(this.getClass))
                    .set("spark.serializer",
"org.apache.spark.serializer.KryoSerializer")
                    //.set("spark.closure.serializer",
"org.apache.spark.serializer.KryoSerializer")
                    .set("spark.kryoserializer.buffer.mb", "256")
                    .set("spark.kryo.registrator",
"com.semi.nlp.MyRegistrator")
                    .set("spark.cores.max", "30")
        new SparkContext(conf)
    }

    def main(args: Array[String]) {
        val spark = initspark("word filter mapping")
        val stopset = spark broadcast
Source.fromURL(this.getClass.getResource("/stoplist.txt")).getLines.map(_.trim).toSet
        val file = spark.textFile("hdfs://ns1/nlp/wiki.splited")
        val tf_map = spark broadcast
file.flatMap(_.split("\t")).map((_,1)).reduceByKey(_+_).countByKey
        val df_map = spark broadcast
file.flatMap(x=>Set(x.split("\t"):_*).toBuffer).map((_,1)).reduceByKey(_+_).countByKey
        val word_mapping = spark broadcast
Map(df_map.value.keys.zipWithIndex.toBuffer:_*)
        def w_filter(w:String) = if (tf_map.value(w) < 8 || df_map.value(w)
< 4 || (stopset.value contains w)) false else true
        val mapped =
file.map(_.split("\t").filter(w_filter).map(w=>word_mapping.value(w)).mkString("\t"))
       
spark.parallelize(word_mapping.value.toSeq).saveAsTextFile("hdfs://ns1/nlp/word_mapping")
        mapped.saveAsTextFile("hdfs://ns1/nlp/lda/wiki.docs")
        spark.stop()
    }
}



--
View this message in context: 
http://apache-spark-user-list.1001560.n3.nabble.com/parallelize-for-a-large-Seq-is-extreamly-slow-tp4801p4871.html
Sent from the Apache Spark User List mailing list archive at Nabble.com.