Re: Docker configuration for akka spark streaming

[David Gomez Saavedra]

The issue is related to this
https://issues.apache.org/jira/browse/SPARK-13906

.set("spark.rpc.netty.dispatcher.numThreads","2")

seem to fix the problem

On Tue, Mar 15, 2016 at 6:45 AM, David Gomez Saavedra 
wrote:

> I have updated the config since I realized the actor system was listening
> on driver port + 1. So changed the ports in my program + the docker images
>
> val conf = new SparkConf()
>   .setMaster(sparkMaster)
>   //.setMaster("local[2]")
>   .setAppName(sparkApp)
>   .set("spark.cassandra.connection.host", CassandraConfig.host)
>   .set("spark.logConf", "true")
>   .set("spark.driver.port","7001")
>   .set("spark.driver.host","192.168.33.10")
>   .set("spark.fileserver.port","6002")
>   .set("spark.broadcast.port","6003")
>   .set("spark.replClassServer.port","6004")
>   .set("spark.blockManager.port","6005")
>   .set("spark.executor.port","6006")
>   
> .set("spark.broadcast.factory","org.apache.spark.broadcast.HttpBroadcastFactory")
>   .setJars(sparkJars)
>
> Netstat of my stream app
>
> tcp6   0  0 :::6002 :::*LISTEN
>  9314/java
> tcp6   0  0 :::6003 :::*LISTEN
>  9314/java
> tcp6   0  0 :::6005 :::*LISTEN
>  9314/java
> tcp6   0  0 192.168.33.10:7001  :::*
>  LISTEN  9314/java
> tcp6   0  0 192.168.33.10:7002  :::*
>  LISTEN  9314/java
> tcp6   0  0 :::4040 :::*LISTEN
>  9314/java
>
> netstat of the master running on docker
>
> Proto Recv-Q Send-Q Local Address   Foreign Address State
>   PID/Program name
> tcp6   0  0 172.18.0.3:7077 :::*
>  LISTEN  -
> tcp6   0  0 :::8080 :::*LISTEN
>  -
> tcp6   0  0 172.18.0.3:6066 :::*
>  LISTEN  -
>
> netstat of worker running on docker
>
> Proto Recv-Q Send-Q Local Address   Foreign Address State
>   PID/Program name
> tcp6   0  0 :::8081 :::*LISTEN
>  -
> tcp6   0  0 :::6005 :::*LISTEN
>  -
> tcp6   0  0 172.18.0.2:6006 :::*
>  LISTEN  -
> tcp6   0  0 172.18.0.2: :::*
>  LISTEN  -
>
>
> so far still no success
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
> On Mon, Mar 14, 2016 at 11:14 PM, Shixiong(Ryan) Zhu <
> shixi...@databricks.com> wrote:
>
>> Could you use netstat to show the ports that the driver is listening?
>>
>> On Mon, Mar 14, 2016 at 1:45 PM, David Gomez Saavedra 
>> wrote:
>>
>>> hi everyone,
>>>
>>> I'm trying to set up spark streaming using akka with a similar example
>>> of the word count provided. When using spark master in local mode
>>> everything works but when I try to run it the driver and executors using
>>> docker I get the following exception
>>>
>>>
>>> 16/03/14 20:32:03 WARN NettyRpcEndpointRef: Error sending message [message 
>>> = Heartbeat(0,[Lscala.Tuple2;@5ad3f40c,BlockManagerId(0, 172.18.0.4, 
>>> 7005))] in 1 attempts
>>> org.apache.spark.rpc.RpcTimeoutException: Cannot receive any reply in 10 
>>> seconds. This timeout is controlled by spark.executor.heartbeatInterval
>>> at 
>>> org.apache.spark.rpc.RpcTimeout.org$apache$spark$rpc$RpcTimeout$$createRpcTimeoutException(RpcTimeout.scala:48)
>>> at 
>>> org.apache.spark.rpc.RpcTimeout$$anonfun$addMessageIfTimeout$1.applyOrElse(RpcTimeout.scala:63)
>>> at 
>>> org.apache.spark.rpc.RpcTimeout$$anonfun$addMessageIfTimeout$1.applyOrElse(RpcTimeout.scala:59)
>>> at 
>>> scala.runtime.AbstractPartialFunction.apply(AbstractPartialFunction.scala:36)
>>> at scala.util.Failure$$anonfun$recover$1.apply(Try.scala:216)
>>> at scala.util.Try$.apply(Try.scala:192)
>>> at scala.util.Failure.recover(Try.scala:216)
>>> at scala.concurrent.Future$$anonfun$recover$1.apply(Future.scala:324)
>>> at scala.concurrent.Future$$anonfun$recover$1.apply(Future.scala:324)
>>> at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32)
>>> at 
>>> org.spark-project.guava.util.concurrent.MoreExecutors$SameThreadExecutorService.execute(MoreExecutors.java:293)
>>> at 
>>> scala.concurrent.impl.ExecutionContextImpl$$anon$1.execute(ExecutionContextImpl.scala:136)
>>> at 
>>> scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:40)
>>> at 
>>> scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:248)
>>> at scala.concurrent.Promise$class.complete(Promise.scala:55)
>>> at 
>>> scala.concurrent.impl.Promise$DefaultPromise.complete(Promise.scala:153)
>>> at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:235)
>>> at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:235)
>>> at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32)
>>> at 
>>> 

Re: Docker configuration for akka spark streaming

[David Gomez Saavedra]

I have updated the config since I realized the actor system was listening
on driver port + 1. So changed the ports in my program + the docker images

val conf = new SparkConf()
  .setMaster(sparkMaster)
  //.setMaster("local[2]")
  .setAppName(sparkApp)
  .set("spark.cassandra.connection.host", CassandraConfig.host)
  .set("spark.logConf", "true")
  .set("spark.driver.port","7001")
  .set("spark.driver.host","192.168.33.10")
  .set("spark.fileserver.port","6002")
  .set("spark.broadcast.port","6003")
  .set("spark.replClassServer.port","6004")
  .set("spark.blockManager.port","6005")
  .set("spark.executor.port","6006")
  
.set("spark.broadcast.factory","org.apache.spark.broadcast.HttpBroadcastFactory")
  .setJars(sparkJars)

Netstat of my stream app

tcp6   0  0 :::6002 :::*LISTEN
 9314/java
tcp6   0  0 :::6003 :::*LISTEN
 9314/java
tcp6   0  0 :::6005 :::*LISTEN
 9314/java
tcp6   0  0 192.168.33.10:7001  :::*LISTEN
 9314/java
tcp6   0  0 192.168.33.10:7002  :::*LISTEN
 9314/java
tcp6   0  0 :::4040 :::*LISTEN
 9314/java

netstat of the master running on docker

Proto Recv-Q Send-Q Local Address   Foreign Address State
PID/Program name
tcp6   0  0 172.18.0.3:7077 :::*LISTEN
 -
tcp6   0  0 :::8080 :::*LISTEN
 -
tcp6   0  0 172.18.0.3:6066 :::*LISTEN
 -

netstat of worker running on docker

Proto Recv-Q Send-Q Local Address   Foreign Address State
PID/Program name
tcp6   0  0 :::8081 :::*LISTEN
 -
tcp6   0  0 :::6005 :::*LISTEN
 -
tcp6   0  0 172.18.0.2:6006 :::*LISTEN
 -
tcp6   0  0 172.18.0.2: :::*LISTEN
 -


so far still no success
















On Mon, Mar 14, 2016 at 11:14 PM, Shixiong(Ryan) Zhu <
shixi...@databricks.com> wrote:

> Could you use netstat to show the ports that the driver is listening?
>
> On Mon, Mar 14, 2016 at 1:45 PM, David Gomez Saavedra 
> wrote:
>
>> hi everyone,
>>
>> I'm trying to set up spark streaming using akka with a similar example of
>> the word count provided. When using spark master in local mode everything
>> works but when I try to run it the driver and executors using docker I get
>> the following exception
>>
>>
>> 16/03/14 20:32:03 WARN NettyRpcEndpointRef: Error sending message [message = 
>> Heartbeat(0,[Lscala.Tuple2;@5ad3f40c,BlockManagerId(0, 172.18.0.4, 7005))] 
>> in 1 attempts
>> org.apache.spark.rpc.RpcTimeoutException: Cannot receive any reply in 10 
>> seconds. This timeout is controlled by spark.executor.heartbeatInterval
>>  at 
>> org.apache.spark.rpc.RpcTimeout.org$apache$spark$rpc$RpcTimeout$$createRpcTimeoutException(RpcTimeout.scala:48)
>>  at 
>> org.apache.spark.rpc.RpcTimeout$$anonfun$addMessageIfTimeout$1.applyOrElse(RpcTimeout.scala:63)
>>  at 
>> org.apache.spark.rpc.RpcTimeout$$anonfun$addMessageIfTimeout$1.applyOrElse(RpcTimeout.scala:59)
>>  at 
>> scala.runtime.AbstractPartialFunction.apply(AbstractPartialFunction.scala:36)
>>  at scala.util.Failure$$anonfun$recover$1.apply(Try.scala:216)
>>  at scala.util.Try$.apply(Try.scala:192)
>>  at scala.util.Failure.recover(Try.scala:216)
>>  at scala.concurrent.Future$$anonfun$recover$1.apply(Future.scala:324)
>>  at scala.concurrent.Future$$anonfun$recover$1.apply(Future.scala:324)
>>  at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32)
>>  at 
>> org.spark-project.guava.util.concurrent.MoreExecutors$SameThreadExecutorService.execute(MoreExecutors.java:293)
>>  at 
>> scala.concurrent.impl.ExecutionContextImpl$$anon$1.execute(ExecutionContextImpl.scala:136)
>>  at 
>> scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:40)
>>  at 
>> scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:248)
>>  at scala.concurrent.Promise$class.complete(Promise.scala:55)
>>  at 
>> scala.concurrent.impl.Promise$DefaultPromise.complete(Promise.scala:153)
>>  at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:235)
>>  at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:235)
>>  at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32)
>>  at 
>> scala.concurrent.BatchingExecutor$Batch$$anonfun$run$1.processBatch$1(BatchingExecutor.scala:63)
>>  at 
>> scala.concurrent.BatchingExecutor$Batch$$anonfun$run$1.apply$mcV$sp(BatchingExecutor.scala:78)
>>  at 
>> 

Re: Docker configuration for akka spark streaming

[Shixiong(Ryan) Zhu]

Could you use netstat to show the ports that the driver is listening?

On Mon, Mar 14, 2016 at 1:45 PM, David Gomez Saavedra 
wrote:

> hi everyone,
>
> I'm trying to set up spark streaming using akka with a similar example of
> the word count provided. When using spark master in local mode everything
> works but when I try to run it the driver and executors using docker I get
> the following exception
>
>
> 16/03/14 20:32:03 WARN NettyRpcEndpointRef: Error sending message [message = 
> Heartbeat(0,[Lscala.Tuple2;@5ad3f40c,BlockManagerId(0, 172.18.0.4, 7005))] in 
> 1 attempts
> org.apache.spark.rpc.RpcTimeoutException: Cannot receive any reply in 10 
> seconds. This timeout is controlled by spark.executor.heartbeatInterval
>   at 
> org.apache.spark.rpc.RpcTimeout.org$apache$spark$rpc$RpcTimeout$$createRpcTimeoutException(RpcTimeout.scala:48)
>   at 
> org.apache.spark.rpc.RpcTimeout$$anonfun$addMessageIfTimeout$1.applyOrElse(RpcTimeout.scala:63)
>   at 
> org.apache.spark.rpc.RpcTimeout$$anonfun$addMessageIfTimeout$1.applyOrElse(RpcTimeout.scala:59)
>   at 
> scala.runtime.AbstractPartialFunction.apply(AbstractPartialFunction.scala:36)
>   at scala.util.Failure$$anonfun$recover$1.apply(Try.scala:216)
>   at scala.util.Try$.apply(Try.scala:192)
>   at scala.util.Failure.recover(Try.scala:216)
>   at scala.concurrent.Future$$anonfun$recover$1.apply(Future.scala:324)
>   at scala.concurrent.Future$$anonfun$recover$1.apply(Future.scala:324)
>   at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32)
>   at 
> org.spark-project.guava.util.concurrent.MoreExecutors$SameThreadExecutorService.execute(MoreExecutors.java:293)
>   at 
> scala.concurrent.impl.ExecutionContextImpl$$anon$1.execute(ExecutionContextImpl.scala:136)
>   at 
> scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:40)
>   at 
> scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:248)
>   at scala.concurrent.Promise$class.complete(Promise.scala:55)
>   at 
> scala.concurrent.impl.Promise$DefaultPromise.complete(Promise.scala:153)
>   at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:235)
>   at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:235)
>   at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32)
>   at 
> scala.concurrent.BatchingExecutor$Batch$$anonfun$run$1.processBatch$1(BatchingExecutor.scala:63)
>   at 
> scala.concurrent.BatchingExecutor$Batch$$anonfun$run$1.apply$mcV$sp(BatchingExecutor.scala:78)
>   at 
> scala.concurrent.BatchingExecutor$Batch$$anonfun$run$1.apply(BatchingExecutor.scala:55)
>   at 
> scala.concurrent.BatchingExecutor$Batch$$anonfun$run$1.apply(BatchingExecutor.scala:55)
>   at 
> scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:72)
>   at 
> scala.concurrent.BatchingExecutor$Batch.run(BatchingExecutor.scala:54)
>   at 
> scala.concurrent.Future$InternalCallbackExecutor$.unbatchedExecute(Future.scala:599)
>   at 
> scala.concurrent.BatchingExecutor$class.execute(BatchingExecutor.scala:106)
>   at 
> scala.concurrent.Future$InternalCallbackExecutor$.execute(Future.scala:597)
>   at 
> scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:40)
>   at 
> scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:248)
>   at scala.concurrent.Promise$class.tryFailure(Promise.scala:112)
>   at 
> scala.concurrent.impl.Promise$DefaultPromise.tryFailure(Promise.scala:153)
>   at 
> org.apache.spark.rpc.netty.NettyRpcEnv$$anon$1.run(NettyRpcEnv.scala:241)
>   at 
> java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
>   at java.util.concurrent.FutureTask.run(FutureTask.java:266)
>   at 
> java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
>   at 
> java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
>   at 
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
>   at 
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
>   at java.lang.Thread.run(Thread.java:745)
> Caused by: java.util.concurrent.TimeoutException: Cannot receive any reply in 
> 10 seconds
>   at 
> org.apache.spark.rpc.netty.NettyRpcEnv$$anon$1.run(NettyRpcEnv.scala:242)
>   ... 7 more
>
>
>
> Here is the config of the spark streaming app
>
> val conf = new SparkConf()
>   .setMaster(sparkMaster)
>   .setAppName(sparkApp)
>   .set("spark.cassandra.connection.host", CassandraConfig.host)
>   .set("spark.logConf", "true")
>   .set("spark.fileserver.port","7002")
>   .set("spark.broadcast.port","7003")
>   .set("spark.replClassServer.port","7004")
>   .set("spark.blockManager.port","7005")
>   .set("spark.executor.port","7006")
>   
> 

Docker configuration for akka spark streaming

[David Gomez Saavedra]

hi everyone,

I'm trying to set up spark streaming using akka with a similar example of
the word count provided. When using spark master in local mode everything
works but when I try to run it the driver and executors using docker I get
the following exception


16/03/14 20:32:03 WARN NettyRpcEndpointRef: Error sending message
[message = Heartbeat(0,[Lscala.Tuple2;@5ad3f40c,BlockManagerId(0,
172.18.0.4, 7005))] in 1 attempts
org.apache.spark.rpc.RpcTimeoutException: Cannot receive any reply in
10 seconds. This timeout is controlled by
spark.executor.heartbeatInterval
at 
org.apache.spark.rpc.RpcTimeout.org$apache$spark$rpc$RpcTimeout$$createRpcTimeoutException(RpcTimeout.scala:48)
at 
org.apache.spark.rpc.RpcTimeout$$anonfun$addMessageIfTimeout$1.applyOrElse(RpcTimeout.scala:63)
at 
org.apache.spark.rpc.RpcTimeout$$anonfun$addMessageIfTimeout$1.applyOrElse(RpcTimeout.scala:59)
at 
scala.runtime.AbstractPartialFunction.apply(AbstractPartialFunction.scala:36)
at scala.util.Failure$$anonfun$recover$1.apply(Try.scala:216)
at scala.util.Try$.apply(Try.scala:192)
at scala.util.Failure.recover(Try.scala:216)
at scala.concurrent.Future$$anonfun$recover$1.apply(Future.scala:324)
at scala.concurrent.Future$$anonfun$recover$1.apply(Future.scala:324)
at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32)
at 
org.spark-project.guava.util.concurrent.MoreExecutors$SameThreadExecutorService.execute(MoreExecutors.java:293)
at 
scala.concurrent.impl.ExecutionContextImpl$$anon$1.execute(ExecutionContextImpl.scala:136)
at 
scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:40)
at 
scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:248)
at scala.concurrent.Promise$class.complete(Promise.scala:55)
at 
scala.concurrent.impl.Promise$DefaultPromise.complete(Promise.scala:153)
at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:235)
at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:235)
at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32)
at 
scala.concurrent.BatchingExecutor$Batch$$anonfun$run$1.processBatch$1(BatchingExecutor.scala:63)
at 
scala.concurrent.BatchingExecutor$Batch$$anonfun$run$1.apply$mcV$sp(BatchingExecutor.scala:78)
at 
scala.concurrent.BatchingExecutor$Batch$$anonfun$run$1.apply(BatchingExecutor.scala:55)
at 
scala.concurrent.BatchingExecutor$Batch$$anonfun$run$1.apply(BatchingExecutor.scala:55)
at 
scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:72)
at 
scala.concurrent.BatchingExecutor$Batch.run(BatchingExecutor.scala:54)
at 
scala.concurrent.Future$InternalCallbackExecutor$.unbatchedExecute(Future.scala:599)
at 
scala.concurrent.BatchingExecutor$class.execute(BatchingExecutor.scala:106)
at 
scala.concurrent.Future$InternalCallbackExecutor$.execute(Future.scala:597)
at 
scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:40)
at 
scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:248)
at scala.concurrent.Promise$class.tryFailure(Promise.scala:112)
at 
scala.concurrent.impl.Promise$DefaultPromise.tryFailure(Promise.scala:153)
at 
org.apache.spark.rpc.netty.NettyRpcEnv$$anon$1.run(NettyRpcEnv.scala:241)
at 
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at 
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at 
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.util.concurrent.TimeoutException: Cannot receive any
reply in 10 seconds
at 
org.apache.spark.rpc.netty.NettyRpcEnv$$anon$1.run(NettyRpcEnv.scala:242)
... 7 more



Here is the config of the spark streaming app

val conf = new SparkConf()
  .setMaster(sparkMaster)
  .setAppName(sparkApp)
  .set("spark.cassandra.connection.host", CassandraConfig.host)
  .set("spark.logConf", "true")
  .set("spark.fileserver.port","7002")
  .set("spark.broadcast.port","7003")
  .set("spark.replClassServer.port","7004")
  .set("spark.blockManager.port","7005")
  .set("spark.executor.port","7006")
  
.set("spark.broadcast.factory","org.apache.spark.broadcast.HttpBroadcastFactory")
  .setJars(sparkJars)

val sc = new SparkContext(conf)

val ssc = new StreamingContext(sc, Seconds(5))

val tags = ssc.actorStream[String](Props(new