ODIn/spark-instrumented-optimizer
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Tuning Akka for more connections

Browse source
This commit is contained in:
root 2012-09-05 07:08:07 +00:00
parent 25fd684b89
commit b7ad291ac5
2 changed files with 6 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
core/src/main/scala/spark/util/AkkaUtils.scala
View file

@ -31,6 +31,7 @@ object AkkaUtils {
akka.remote.netty.hostname = "%s"
akka.remote.netty.port = %d
akka.remote.netty.connection-timeout = 1s
akka.remote.netty.execution-pool-size = 10
""".format(host, port))
val actorSystem = ActorSystem("spark", akkaConf, getClass.getClassLoader)

10
streaming/src/main/scala/spark/streaming/examples/WordCount2.scala
View file

@ -62,10 +62,10 @@ object WordCount2_ExtraFunctions {
object WordCount2 {
def warmup(sc: SparkContext) {
(0 until 10).foreach {i =>
sc.parallelize(1 to 20000000, 1000)
(0 until 3).foreach {i =>
sc.parallelize(1 to 20000000, 500)
.map(x => (x % 337, x % 1331))
.reduceByKey(_ + _)
.reduceByKey(_ + _, 100)
.count()
}
}
@ -84,11 +84,11 @@ object WordCount2 {
val ssc = new StreamingContext(master, "WordCount2")
ssc.setBatchDuration(batchDuration)
//warmup(ssc.sc)
warmup(ssc.sc)
val data = ssc.sc.textFile(file, mapTasks.toInt).persist(
new StorageLevel(false, true, false, 3)) // Memory only, serialized, 3 replicas
println("Data count: " + data.count())
println("Data count: " + data.map(x => if (x == "") 1 else x.split(" ").size / x.split(" ").size).count())
println("Data count: " + data.count())
println("Data count: " + data.count())