Please see the comments at the tail of SPARK-2356
Cheers
On Wed, Aug 5, 2015 at 6:04 PM, Ashish Dutt <ashish.du...@gmail.com> wrote:
> *Use Case:* To automate the process of data extraction (HDFS), data
> analysis (pySpark/sparkR) and saving the data back to HDFS
> programmatically.
>
> *Prospective solutions:*
>
> 1. Create a remote server connectivity program in an IDE like pyCharm or
> RStudio and use it to retrieve the data from HDFS or else
> 2. Create the data retrieval code in python or R and then point the IDE to
> the remote server using TCP.
>
> *Problem:* How to achieve either of the prospective solution 1 or 2
> defined above? Do you have any better solution then these, if yes please
> share?
>
> *What have I tried so far?*
>
> The server and 3 namenodes already installed with pyspark and I have
> checked pyspark works in standalone mode on all four servers. Pyspark works
> in standalone mode on my laptop too.
>
> I use the following code but I am not able to connect to the remote server.
>
> import os
> import sys
> try:
> from pyspark import SparkContext
> from pyspark import SparkConf
> print ("Pyspark sucess")
> except ImportError as e:
> print ("Error importing Spark Modules", e)
>
> conf = SparkConf()
> conf.setMaster("spark://10.210.250.400:7077")
> conf.setAppName("First_Remote_Spark_Program")
> sc = SparkContext(conf=conf)
> print ("connection succeeded with Master",conf)
> data = [1, 2, 3, 4, 5]
> distData = sc.parallelize(data)
> print(distData)
>
> The stack trace of error is
>
> Pyspark sucess
> 15/08/01 14:08:24 WARN NativeCodeLoader: Unable to load native-hadoop library
> for your platform... using builtin-java classes where applicable
> 15/08/01 14:08:24 ERROR Shell: Failed to locate the winutils binary in the
> hadoop binary path
> java.io.IOException: Could not locate executable null\bin\winutils.exe in the
> Hadoop binaries.
> at org.apache.hadoop.util.Shell.getQualifiedBinPath(Shell.java:318)
> at org.apache.hadoop.util.Shell.getWinUtilsPath(Shell.java:333)
> at org.apache.hadoop.util.Shell.<clinit>(Shell.java:326)
> at org.apache.hadoop.util.StringUtils.<clinit>(StringUtils.java:76)
> at org.apache.hadoop.security.Groups.parseStaticMapping(Groups.java:93)
> at org.apache.hadoop.security.Groups.<init>(Groups.java:77)
> at
> org.apache.hadoop.security.Groups.getUserToGroupsMappingService(Groups.java:240)
> at
> org.apache.hadoop.security.UserGroupInformation.initialize(UserGroupInformation.java:255)
> at
> org.apache.hadoop.security.UserGroupInformation.ensureInitialized(UserGroupInformation.java:232)
> at
> org.apache.hadoop.security.UserGroupInformation.loginUserFromSubject(UserGroupInformation.java:718)
> at
> org.apache.hadoop.security.UserGroupInformation.getLoginUser(UserGroupInformation.java:703)
> at
> org.apache.hadoop.security.UserGroupInformation.getCurrentUser(UserGroupInformation.java:605)
> at
> org.apache.spark.util.Utils$$anonfun$getCurrentUserName$1.apply(Utils.scala:2162)
> at
> org.apache.spark.util.Utils$$anonfun$getCurrentUserName$1.apply(Utils.scala:2162)
> at scala.Option.getOrElse(Option.scala:120)
> at org.apache.spark.util.Utils$.getCurrentUserName(Utils.scala:2162)
> at org.apache.spark.SparkContext.<init>(SparkContext.scala:301)
> at
> org.apache.spark.api.java.JavaSparkContext.<init>(JavaSparkContext.scala:61)
> at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
> at
> sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:57)
> at
> sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
> at java.lang.reflect.Constructor.newInstance(Constructor.java:526)
> at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:234)
> at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:379)
> at py4j.Gateway.invoke(Gateway.java:214)
> at
> py4j.commands.ConstructorCommand.invokeConstructor(ConstructorCommand.java:79)
> at py4j.commands.ConstructorCommand.execute(ConstructorCommand.java:68)
> at py4j.GatewayConnection.run(GatewayConnection.java:207)
> at java.lang.Thread.run(Thread.java:745)
> 15/08/01 14:08:25 WARN Utils: Service 'SparkUI' could not bind on port 4040.
> Attempting port 4041.
> 15/08/01 14:08:26 WARN AppClient$ClientActor: Could not connect to
> akka.tcp://sparkMaster@10.210.250.400:7077: akka.remote.InvalidAssociation:
> Invalid address: akka.tcp://sparkMaster@10.210.250.400:7077
> 15/08/01 14:08:26 WARN Remoting: Tried to associate with unreachable remote
> address [akka.tcp://sparkMaster@10.210.250.400:7077]. Address is now gated
> for 5000 ms, all messages to this address will be delivered to dead letters.
> Reason: Connection refused: no further information: /10.210.250.400:7077
> 15/08/01 14:08:46 WARN AppClient$ClientActor: Could not connect to
> akka.tcp://sparkMaster@10.210.250.400:7077: akka.remote.InvalidAssociation:
> Invalid address: akka.tcp://sparkMaster@10.210.250.400:7077
> 15/08/01 14:08:46 WARN Remoting: Tried to associate with unreachable remote
> address [akka.tcp://sparkMaster@10.210.250.400:7077]. Address is now gated
> for 5000 ms, all messages to this address will be delivered to dead letters.
> Reason: Connection refused: no further information: /10.210.250.400:7077
> 15/08/01 14:09:06 WARN AppClient$ClientActor: Could not connect to
> akka.tcp://sparkMaster@10.210.250.400:7077: akka.remote.InvalidAssociation:
> Invalid address: akka.tcp://sparkMaster@10.210.250.400:7077
> 15/08/01 14:09:06 WARN Remoting: Tried to associate with unreachable remote
> address [akka.tcp://sparkMaster@10.210.250.400:7077]. Address is now gated
> for 5000 ms, all messages to this address will be delivered to dead letters.
> Reason: Connection refused: no further information: /10.210.250.400:7077
> 15/08/01 14:09:25 ERROR SparkDeploySchedulerBackend: Application has been
> killed. Reason: All masters are unresponsive! Giving up.
> 15/08/01 14:09:25 WARN SparkDeploySchedulerBackend: Application ID is not
> initialized yet.
> 15/08/01 14:09:25 ERROR OneForOneStrategy:
> java.lang.NullPointerException
> at
> org.apache.spark.deploy.client.AppClient$ClientActor$$anonfun$receiveWithLogging$1.applyOrElse(AppClient.scala:160)
> at
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
> at
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
> at
> scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
> at
> org.apache.spark.util.ActorLogReceive$$anon$1.apply(ActorLogReceive.scala:59)
> at
> org.apache.spark.util.ActorLogReceive$$anon$1.apply(ActorLogReceive.scala:42)
> at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:118)
> at
> org.apache.spark.util.ActorLogReceive$$anon$1.applyOrElse(ActorLogReceive.scala:42)
> at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
> at
> org.apache.spark.deploy.client.AppClient$ClientActor.aroundReceive(AppClient.scala:61)
> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
> at akka.actor.ActorCell.invoke(ActorCell.scala:487)
> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:238)
> at akka.dispatch.Mailbox.run(Mailbox.scala:220)
> at
> akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:393)
> at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
> at
> scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
> at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
> at
> scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
> 15/08/01 14:09:25 ERROR SparkContext: Error initializing SparkContext.
> java.lang.IllegalStateException: Cannot call methods on a stopped SparkContext
> at org.apache.spark.SparkContext.org
> <http://org.apache.spark.sparkcontext.org/>$apache$spark$SparkContext$$assertNotStopped(SparkContext.scala:103)
> at org.apache.spark.SparkContext.getSchedulingMode(SparkContext.scala:1501)
> at
> org.apache.spark.SparkContext.postEnvironmentUpdate(SparkContext.scala:2005)
> at org.apache.spark.SparkContext.<init>(SparkContext.scala:543)
> at
> org.apache.spark.api.java.JavaSparkContext.<init>(JavaSparkContext.scala:61)
> at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
> at
> sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:57)
> at
> sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
> at java.lang.reflect.Constructor.newInstance(Constructor.java:526)
> at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:234)
> at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:379)
> at py4j.Gateway.invoke(Gateway.java:214)
> at
> py4j.commands.ConstructorCommand.invokeConstructor(ConstructorCommand.java:79)
> at py4j.commands.ConstructorCommand.execute(ConstructorCommand.java:68)
> at py4j.GatewayConnection.run(GatewayConnection.java:207)
> at java.lang.Thread.run(Thread.java:745)
> Traceback (most recent call last):
> File "C:/Users/ashish dutt/PycharmProjects/KafkaToHDFS/local2Remote.py", line
> 26, in <module>
> sc = SparkContext(conf=conf)
> File "C:\spark-1.4.0\python\pyspark\context.py", line 113, in __init__
> conf, jsc, profiler_cls)
> File "C:\spark-1.4.0\python\pyspark\context.py", line 165, in _do_init
> self._jsc = jsc or self._initialize_context(self._conf._jconf)
> File "C:\spark-1.4.0\python\pyspark\context.py", line 219, in
> _initialize_context
> return self._jvm.JavaSparkContext(jconf)
> File "C:\spark-1.4.0\python\lib\py4j-0.8.2.1-src.zip\py4j\java_gateway.py",
> line 701, in __call__
> File "C:\spark-1.4.0\python\lib\py4j-0.8.2.1-src.zip\py4j\protocol.py", line
> 300, in get_return_value
> py4j.protocol.Py4JJavaError: An error occurred while calling
> None.org.apache.spark.api.java.JavaSparkContext.
> : java.lang.IllegalStateException: Cannot call methods on a stopped
> SparkContext
> at org.apache.spark.SparkContext.org
> <http://org.apache.spark.sparkcontext.org/>$apache$spark$SparkContext$$assertNotStopped(SparkContext.scala:103)
> at org.apache.spark.SparkContext.getSchedulingMode(SparkContext.scala:1501)
> at
> org.apache.spark.SparkContext.postEnvironmentUpdate(SparkContext.scala:2005)
> at org.apache.spark.SparkContext.<init>(SparkContext.scala:543)
> at
> org.apache.spark.api.java.JavaSparkContext.<init>(JavaSparkContext.scala:61)
> at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
> at
> sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:57)
> at
> sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
> at java.lang.reflect.Constructor.newInstance(Constructor.java:526)
> at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:234)
> at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:379)
> at py4j.Gateway.invoke(Gateway.java:214)
> at
> py4j.commands.ConstructorCommand.invokeConstructor(ConstructorCommand.java:79)
> at py4j.commands.ConstructorCommand.execute(ConstructorCommand.java:68)
> at py4j.GatewayConnection.run(GatewayConnection.java:207)
> at java.lang.Thread.run(Thread.java:745)
>
> Process finished with exit code 1
>
> The spark-defaults.conf file is configured as follows
>
> #spark.eventLog.dir=hdfs://ABCD01:8020/user/spark/applicationHistory
> spark.eventLog.dir hdfs://10.210.250.400:8020/user/spark/eventlog
> spark.eventLog.enabled true
> spark.serializer org.apache.spark.serializer.KryoSerializer
> spark.shuffle.service.enabled true
> spark.shuffle.service.port 7337
> spark.yarn.historyServer.address http://ABCD04:18088 <http://abcd04:18088/>
> spark.master spark://10.210.250.400:7077
> spark.yarn.jar
> local:/opt/cloudera/parcels/CDH-5.4.2-1.cdh5.4.2.p0.2/lib/spark/assembly/lib/spark-assembly-1.3.0-cdh5.4.2-hadoop2.6.0-cdh5.4.2.jar
> spark.driver.extraLibraryPath
> /opt/cloudera/parcels/CDH-5.4.2-1.cdh5.4.2.p0.2/lib/hadoop/lib/native
> spark.executor.extraLibraryPath
> /opt/cloudera/parcels/CDH-5.4.2-1.cdh5.4.2.p0.2/lib/hadoop/lib/native
> spark.yarn.am.extraLibraryPath
> /opt/cloudera/parcels/CDH-5.4.2-1.cdh5.4.2.p0.2/lib/hadoop/lib/native
> spark.logConf true
>
> The spark-env.sh file is configured as follows
>
> #!/usr/bin/env bash
> ##
> # Generated by Cloudera Manager and should not be modified directly
> ##
>
> SELF="$(cd $(dirname $BASH_SOURCE) && pwd)"
> if [ -z "$SPARK_CONF_DIR" ]; then
> export SPARK_CONF_DIR="$SELF"
> fi
>
> export SPARK_HOME=/opt/cloudera/parcels/CDH-5.4.2-1.cdh5.4.2.p0.2/lib/spark
> export
> DEFAULT_HADOOP_HOME=/opt/cloudera/parcels/CDH-5.4.2-1.cdh5.4.2.p0.2/lib/hadoop
> #export STANDALONE_SPARK_MASTER_HOST=`ABCD01`
> export SPARK_MASTER_IP=spark://10.210.250.400
> export SPARK_MASTER_PORT=7077
> export SPARK_WEBUI_PORT=18080
>
>
> ### Path of Spark assembly jar in HDFS
> export SPARK_JAR_HDFS_PATH=${SPARK_JAR_HDFS_PATH:-''}
>
> export HADOOP_HOME=${HADOOP_HOME:-$DEFAULT_HADOOP_HOME}
>
> if [ -n "$HADOOP_HOME" ]; then
> LD_LIBRARY_PATH=$LD_LIBRARY_PATH:${HADOOP_HOME}/lib/native
> fi
>
> SPARK_EXTRA_LIB_PATH=""
> if [ -n "$SPARK_EXTRA_LIB_PATH" ]; then
> LD_LIBRARY_PATH=$LD_LIBRARY_PATH:$SPARK_EXTRA_LIB_PATH
> fi
>
> export LD_LIBRARY_PATH
> export HADOOP_CONF_DIR=${HADOOP_CONF_DIR:-$SPARK_CONF_DIR/yarn-conf}
>
> # This is needed to support old CDH versions that use a forked version
> # of compute-classpath.sh.
> export SCALA_LIBRARY_PATH=${SPARK_HOME}/lib
>
> # Set distribution classpath. This is only used in CDH 5.3 and later.
> export SPARK_DIST_CLASSPATH=$(paste -sd: "$SELF/classpath.txt")
>
> And the slaves.sh file is configured as
>
> 10.210.250.401
> 10.210.250.402
> 10.210.250.403
>
> Please tell me how can I connect to the remote server using pycharm or any
> other IDE?
>
> Thank you,
>
> Ashish
>
>
