You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@spark.apache.org by sid <si...@gmail.com> on 2014/10/27 07:57:01 UTC
Python code crashing on ReduceByKey if I return custom class object
Hi , I am new to spark and I am trying to use pyspark.
I am trying to find mean of 128 dimension vectors present in a file .
Below is the code
----------------------------------------------------------------
from cStringIO import StringIO
class testing:
def __str__(self):
file_str = StringIO()
file_str.write(str(self.first))
file_str.write(" ")
for n in self.vector:
file_str.write(str(n))
file_str.write(" ")
file_str.write(self.filename)
return file_str.getvalue()
def __init__(self,txt=""):
self.vector = [0.0]*128
if len(txt)==0:
return
i=0
for n in txt.split():
if i==0 :
self.key = float(n)
i = i+1
continue
if i<128:
self.vector[i-1]=float(n)
i = i+1
continue
break
def addVec(self,r):
a = testing()
for n in xrange(0,128):
a.vector[n] = self.vector[n] + r.vector[n]
return a
def InitializeAndReturnPair(string):
vec = testing(string)
return vec.key,vec
from pyspark import SparkConf, SparkContext
conf = (SparkConf()
.setMaster("local")
.setAppName("My app")
.set("spark.executor.memory", "1g"))
sc = SparkContext(conf = conf)
inp = sc.textFile("input.txt")
output = inp.map(lambda s: InitializeAndReturnPair(s)).cache()
#Below line is throwing error
print output.reduceByKey(lambda a,b : addVec(a,b)).collect()
---------------------------------------------
I am not able to figure out where I am missing out , I tried changing the
serializer but still getting similar error.
Error
Spark assembly has been built with Hive, including Datanucleus jars on
classpath Using Spark's default log4j profile:
org/apache/spark/log4j-defaults.properties 14/10/25 05:02:39 WARN Utils:
Your hostname, Sid resolves to a loopback address: 127.0.1.1; using
192.168.0.15 instead (on interface wlan0) 14/10/25 05:02:39 WARN Utils: Set
SPARK_LOCAL_IP if you need to bind to another address 14/10/25 05:02:39 INFO
SecurityManager: Changing view acls to: sid, 14/10/25 05:02:39 INFO
SecurityManager: Changing modify acls to: sid, 14/10/25 05:02:39 INFO
SecurityManager: SecurityManager: authentication disabled; ui acls disabled;
users with view permissions: Set(sid, ); users with modify permissions:
Set(sid, ) 14/10/25 05:02:40 INFO Slf4jLogger: Slf4jLogger started 14/10/25
05:02:40 INFO Remoting: Starting remoting 14/10/25 05:02:40 INFO Remoting:
Remoting started; listening on addresses
:[akka.tcp://sparkDriver@192.168.0.15:52124] 14/10/25 05:02:40 INFO
Remoting: Remoting now listens on addresses:
[akka.tcp://sparkDriver@192.168.0.15:52124] 14/10/25 05:02:40 INFO Utils:
Successfully started service 'sparkDriver' on port 52124. 14/10/25 05:02:40
INFO SparkEnv: Registering MapOutputTracker 14/10/25 05:02:40 INFO SparkEnv:
Registering BlockManagerMaster 14/10/25 05:02:40 INFO DiskBlockManager:
Created local directory at /tmp/spark-local-20141025050240-56c5 14/10/25
05:02:40 INFO Utils: Successfully started service 'Connection manager for
block manager' on port 51705. 14/10/25 05:02:40 INFO ConnectionManager:
Bound socket to port 51705 with id = ConnectionManagerId(192.168.0.15,51705)
14/10/25 05:02:40 INFO MemoryStore: MemoryStore started with capacity 265.4
MB 14/10/25 05:02:40 INFO BlockManagerMaster: Trying to register
BlockManager 14/10/25 05:02:40 INFO BlockManagerMasterActor: Registering
block manager 192.168.0.15:51705 with 265.4 MB RAM 14/10/25 05:02:40 INFO
BlockManagerMaster: Registered BlockManager 14/10/25 05:02:40 INFO
HttpFileServer: HTTP File server directory is
/tmp/spark-ce172226-4732-4633-aa45-1cbd45b7ec98 14/10/25 05:02:40 INFO
HttpServer: Starting HTTP Server 14/10/25 05:02:40 INFO Utils: Successfully
started service 'HTTP file server' on port 55111. 14/10/25 05:02:40 INFO
Utils: Successfully started service 'SparkUI' on port 4040. 14/10/25
05:02:40 INFO SparkUI: Started SparkUI at http ://192.168.0.15:4040 14/10/25
05:02:41 WARN NativeCodeLoader: Unable to load native-hadoop library for
your platform... using builtin-java classes where applicable 14/10/25
05:02:41 INFO Utils: Copying /home/sid/Downloads/spark/pdsWork/AskHelp.py to
/tmp/spark-80005f48-e41d-48ff-b249-bdf87130de5d/AskHelp.py 14/10/25 05:02:41
INFO SparkContext: Added file
file:/home/sid/Downloads/spark/pdsWork/AskHelp.py at http
://192.168.0.15:55111/files/AskHelp.py with timestamp 1414227761547 14/10/25
05:02:41 INFO AkkaUtils: Connecting to HeartbeatReceiver:
akka.tcp://sparkDriver@192.168.0.15:52124/user/HeartbeatReceiver 14/10/25
05:02:41 INFO MemoryStore: ensureFreeSpace(163705) called with curMem=0,
maxMem=278302556 14/10/25 05:02:41 INFO MemoryStore: Block broadcast_0
stored as values in memory (estimated size 159.9 KB, free 265.3 MB) 14/10/25
05:02:42 INFO FileInputFormat: Total input paths to process : 1 14/10/25
05:02:42 INFO SparkContext: Starting job: collect at
/home/sid/Downloads/spark/pdsWork/AskHelp.py:48 14/10/25 05:02:42 INFO
DAGScheduler: Registering RDD 4 (RDD at PythonRDD.scala:261) 14/10/25
05:02:42 INFO DAGScheduler: Got job 0 (collect at
/home/sid/Downloads/spark/pdsWork/AskHelp.py:48) with 1 output partitions
(allowLocal=false) 14/10/25 05:02:42 INFO DAGScheduler: Final stage: Stage
0(collect at /home/sid/Downloads/spark/pdsWork/AskHelp.py:48) 14/10/25
05:02:42 INFO DAGScheduler: Parents of final stage: List(Stage 1) 14/10/25
05:02:42 INFO DAGScheduler: Missing parents: List(Stage 1) 14/10/25 05:02:42
INFO DAGScheduler: Submitting Stage 1 (PairwiseRDD[4] at RDD at
PythonRDD.scala:261), which has no missing parents 14/10/25 05:02:42 INFO
MemoryStore: ensureFreeSpace(9880) called with curMem=163705,
maxMem=278302556 14/10/25 05:02:42 INFO MemoryStore: Block broadcast_1
stored as values in memory (estimated size 9.6 KB, free 265.2 MB) 14/10/25
05:02:42 INFO DAGScheduler: Submitting 1 missing tasks from Stage 1
(PairwiseRDD[4] at RDD at PythonRDD.scala:261) 14/10/25 05:02:42 INFO
TaskSchedulerImpl: Adding task set 1.0 with 1 tasks 14/10/25 05:02:42 INFO
TaskSetManager: Starting task 0.0 in stage 1.0 (TID 0, localhost,
PROCESS_LOCAL, 1245 bytes) 14/10/25 05:02:42 INFO Executor: Running task 0.0
in stage 1.0 (TID 0) 14/10/25 05:02:42 INFO Executor: Fetching http
://192.168.0.15:55111/files/AskHelp.py with timestamp 1414227761547 14/10/25
05:02:42 INFO Utils: Fetching http ://192.168.0.15:55111/files/AskHelp.py to
/tmp/fetchFileTemp2903133416780112069.tmp 14/10/25 05:02:42 INFO
CacheManager: Partition rdd_2_0 not found, computing it 14/10/25 05:02:42
INFO HadoopRDD: Input split:
file:/home/sid/Downloads/spark/pdsWork/input.txt:0+147355 14/10/25 05:02:42
INFO deprecation: mapred.tip.id is deprecated. Instead, use
mapreduce.task.id 14/10/25 05:02:42 INFO deprecation: mapred.task.id is
deprecated. Instead, use mapreduce.task.attempt.id 14/10/25 05:02:42 INFO
deprecation: mapred.task.is.map is deprecated. Instead, use
mapreduce.task.ismap 14/10/25 05:02:42 INFO deprecation:
mapred.task.partition is deprecated. Instead, use mapreduce.task.partition
14/10/25 05:02:42 INFO deprecation: mapred.job.id is deprecated. Instead,
use mapreduce.job.id 14/10/25 05:02:42 ERROR PythonRDD: Python worker exited
unexpectedly (crashed) org.apache.spark.api.python.PythonException:
Traceback (most recent call last): File
"/home/sid/Downloads/spark/python/pyspark/worker.py", line 75, in main
command = pickleSer._read_with_length(infile) File
"/home/sid/Downloads/spark/python/pyspark/serializers.py", line 146, in
_read_with_length length = read_int(stream) File
"/home/sid/Downloads/spark/python/pyspark/serializers.py", line 464, in
read_int raise EOFError EOFError
--
View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/Python-code-crashing-on-ReduceByKey-if-I-return-custom-class-object-tp17317.html
Sent from the Apache Spark User List mailing list archive at Nabble.com.
---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
For additional commands, e-mail: user-help@spark.apache.org
Re: Python code crashing on ReduceByKey if I return custom class
object
Posted by Gen <ge...@gmail.com>.
Hi,
I think the problem is caused by Data Serialization. You can follow the link
https://spark.apache.org/docs/latest/tuning.html
<https://spark.apache.org/docs/latest/tuning.html> to register your class
testing.
For pyspark 1.1.0, there is a problem about the default serializer.
https://issues.apache.org/jira/browse/SPARK-2652?filter=-2
<https://issues.apache.org/jira/browse/SPARK-2652?filter=-2> .
Cheers.
Gen
sid wrote
> Hi , I am new to spark and I am trying to use pyspark.
>
> I am trying to find mean of 128 dimension vectors present in a file .
>
> Below is the code
> ----------------------------------------------------------------
> from cStringIO import StringIO
>
> class testing:
> def __str__(self):
> file_str = StringIO()
> for n in self.vector:
> file_str.write(str(n))
> file_str.write(" ")
> return file_str.getvalue()
> def __init__(self,txt="",initial=False):
> self.vector = [0.0]*128
> if len(txt)==0:
> return
> i=0
> for n in txt.split():
> if i<128:
> self.vector[i]=float(n)
> i = i+1
> continue
> self.filename=n
> break
> def addVec(self,r):
> a = testing()
> for n in xrange(0,128):
> a.vector[n] = self.vector[n] + r.vector[n]
> return a
>
> def InitializeAndReturnPair(string,first=False):
> vec = testing(string,first)
> return 1,vec
>
>
> from pyspark import SparkConf, SparkContext
> conf = (SparkConf()
> .setMaster("local")
> .setAppName("My app")
> .set("spark.executor.memory", "1g"))
> sc = SparkContext(conf = conf)
>
> inp = sc.textFile("input.txt")
> output = inp.map(lambda s: InitializeAndReturnPair(s,True)).cache()
> output.saveAsTextFile("output")
> print output.reduceByKey(lambda a,b : a).collect()
> ---------------------------------------------
>
> Example line in input.txt
> 6.0 156.0 26.0 3.0 1.0 0.0 2.0 1.0 15.0 113.0 53.0 139.0 156.0 0.0 0.0 0.0
> 156.0 29.0 1.0 38.0 59.0 0.0 0.0 0.0 28.0 4.0 2.0 9.0 1.0 0.0 0.0 0.0 9.0
> 83.0 13.0 1.0 0.0 9.0 42.0 7.0 41.0 71.0 74.0 123.0 35.0 17.0 7.0 2.0
> 156.0 27.0 6.0 33.0 11.0 2.0 0.0 11.0 35.0 4.0 2.0 4.0 1.0 3.0 2.0 4.0 0.0
> 0.0 0.0 0.0 2.0 19.0 45.0 17.0 47.0 2.0 2.0 7.0 59.0 90.0 15.0 11.0 156.0
> 14.0 1.0 4.0 9.0 11.0 2.0 29.0 35.0 6.0 5.0 9.0 4.0 2.0 1.0 3.0 1.0 0.0
> 0.0 0.0 1.0 5.0 25.0 14.0 27.0 2.0 0.0 2.0 86.0 48.0 10.0 6.0 156.0 23.0
> 1.0 2.0 21.0 6.0 0.0 3.0 31.0 10.0 4.0 3.0 0.0 0.0 1.0 2.0
>
> I am not able to figure out where I am missing out , I tried changing the
> serializer but still getting similar error.
>
> Place the error here http://pastebin.com/0tqiiJQm
--
View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/Python-code-crashing-on-ReduceByKey-if-I-return-custom-class-object-tp17317p17345.html
Sent from the Apache Spark User List mailing list archive at Nabble.com.
---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
For additional commands, e-mail: user-help@spark.apache.org
Re: Python code crashing on ReduceByKey if I return custom class
object
Posted by sid <si...@gmail.com>.
I have implemented the advise and now the issue is resolved .
--
View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/Python-code-crashing-on-ReduceByKey-if-I-return-custom-class-object-tp17317p17426.html
Sent from the Apache Spark User List mailing list archive at Nabble.com.
---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
For additional commands, e-mail: user-help@spark.apache.org
Re: Python code crashing on ReduceByKey if I return custom class object
Posted by Davies Liu <da...@databricks.com>.
This is known issue with PySpark, the class and objects of custom
class in current script can not serialized by pickle between driver
and worker
You can workaround this by put 'testing' in a module, and sending this
module to cluster by `sc.addPyFile`
Davies
On Sun, Oct 26, 2014 at 11:57 PM, sid <si...@gmail.com> wrote:
> Hi , I am new to spark and I am trying to use pyspark.
>
> I am trying to find mean of 128 dimension vectors present in a file .
>
> Below is the code
> ----------------------------------------------------------------
>
> from cStringIO import StringIO
>
> class testing:
> def __str__(self):
> file_str = StringIO()
> file_str.write(str(self.first))
> file_str.write(" ")
> for n in self.vector:
> file_str.write(str(n))
> file_str.write(" ")
> file_str.write(self.filename)
> return file_str.getvalue()
> def __init__(self,txt=""):
> self.vector = [0.0]*128
> if len(txt)==0:
> return
> i=0
> for n in txt.split():
> if i==0 :
> self.key = float(n)
> i = i+1
> continue
> if i<128:
> self.vector[i-1]=float(n)
> i = i+1
> continue
> break
> def addVec(self,r):
> a = testing()
> for n in xrange(0,128):
> a.vector[n] = self.vector[n] + r.vector[n]
> return a
>
> def InitializeAndReturnPair(string):
> vec = testing(string)
> return vec.key,vec
>
>
> from pyspark import SparkConf, SparkContext
> conf = (SparkConf()
> .setMaster("local")
> .setAppName("My app")
> .set("spark.executor.memory", "1g"))
> sc = SparkContext(conf = conf)
>
> inp = sc.textFile("input.txt")
> output = inp.map(lambda s: InitializeAndReturnPair(s)).cache()
> #Below line is throwing error
> print output.reduceByKey(lambda a,b : addVec(a,b)).collect()
>
> ---------------------------------------------
>
> I am not able to figure out where I am missing out , I tried changing the
> serializer but still getting similar error.
>
> Error
>
> Spark assembly has been built with Hive, including Datanucleus jars on
> classpath Using Spark's default log4j profile:
> org/apache/spark/log4j-defaults.properties 14/10/25 05:02:39 WARN Utils:
> Your hostname, Sid resolves to a loopback address: 127.0.1.1; using
> 192.168.0.15 instead (on interface wlan0) 14/10/25 05:02:39 WARN Utils: Set
> SPARK_LOCAL_IP if you need to bind to another address 14/10/25 05:02:39 INFO
> SecurityManager: Changing view acls to: sid, 14/10/25 05:02:39 INFO
> SecurityManager: Changing modify acls to: sid, 14/10/25 05:02:39 INFO
> SecurityManager: SecurityManager: authentication disabled; ui acls disabled;
> users with view permissions: Set(sid, ); users with modify permissions:
> Set(sid, ) 14/10/25 05:02:40 INFO Slf4jLogger: Slf4jLogger started 14/10/25
> 05:02:40 INFO Remoting: Starting remoting 14/10/25 05:02:40 INFO Remoting:
> Remoting started; listening on addresses
> :[akka.tcp://sparkDriver@192.168.0.15:52124] 14/10/25 05:02:40 INFO
> Remoting: Remoting now listens on addresses:
> [akka.tcp://sparkDriver@192.168.0.15:52124] 14/10/25 05:02:40 INFO Utils:
> Successfully started service 'sparkDriver' on port 52124. 14/10/25 05:02:40
> INFO SparkEnv: Registering MapOutputTracker 14/10/25 05:02:40 INFO SparkEnv:
> Registering BlockManagerMaster 14/10/25 05:02:40 INFO DiskBlockManager:
> Created local directory at /tmp/spark-local-20141025050240-56c5 14/10/25
> 05:02:40 INFO Utils: Successfully started service 'Connection manager for
> block manager' on port 51705. 14/10/25 05:02:40 INFO ConnectionManager:
> Bound socket to port 51705 with id = ConnectionManagerId(192.168.0.15,51705)
> 14/10/25 05:02:40 INFO MemoryStore: MemoryStore started with capacity 265.4
> MB 14/10/25 05:02:40 INFO BlockManagerMaster: Trying to register
> BlockManager 14/10/25 05:02:40 INFO BlockManagerMasterActor: Registering
> block manager 192.168.0.15:51705 with 265.4 MB RAM 14/10/25 05:02:40 INFO
> BlockManagerMaster: Registered BlockManager 14/10/25 05:02:40 INFO
> HttpFileServer: HTTP File server directory is
> /tmp/spark-ce172226-4732-4633-aa45-1cbd45b7ec98 14/10/25 05:02:40 INFO
> HttpServer: Starting HTTP Server 14/10/25 05:02:40 INFO Utils: Successfully
> started service 'HTTP file server' on port 55111. 14/10/25 05:02:40 INFO
> Utils: Successfully started service 'SparkUI' on port 4040. 14/10/25
> 05:02:40 INFO SparkUI: Started SparkUI at http ://192.168.0.15:4040 14/10/25
> 05:02:41 WARN NativeCodeLoader: Unable to load native-hadoop library for
> your platform... using builtin-java classes where applicable 14/10/25
> 05:02:41 INFO Utils: Copying /home/sid/Downloads/spark/pdsWork/AskHelp.py to
> /tmp/spark-80005f48-e41d-48ff-b249-bdf87130de5d/AskHelp.py 14/10/25 05:02:41
> INFO SparkContext: Added file
> file:/home/sid/Downloads/spark/pdsWork/AskHelp.py at http
> ://192.168.0.15:55111/files/AskHelp.py with timestamp 1414227761547 14/10/25
> 05:02:41 INFO AkkaUtils: Connecting to HeartbeatReceiver:
> akka.tcp://sparkDriver@192.168.0.15:52124/user/HeartbeatReceiver 14/10/25
> 05:02:41 INFO MemoryStore: ensureFreeSpace(163705) called with curMem=0,
> maxMem=278302556 14/10/25 05:02:41 INFO MemoryStore: Block broadcast_0
> stored as values in memory (estimated size 159.9 KB, free 265.3 MB) 14/10/25
> 05:02:42 INFO FileInputFormat: Total input paths to process : 1 14/10/25
> 05:02:42 INFO SparkContext: Starting job: collect at
> /home/sid/Downloads/spark/pdsWork/AskHelp.py:48 14/10/25 05:02:42 INFO
> DAGScheduler: Registering RDD 4 (RDD at PythonRDD.scala:261) 14/10/25
> 05:02:42 INFO DAGScheduler: Got job 0 (collect at
> /home/sid/Downloads/spark/pdsWork/AskHelp.py:48) with 1 output partitions
> (allowLocal=false) 14/10/25 05:02:42 INFO DAGScheduler: Final stage: Stage
> 0(collect at /home/sid/Downloads/spark/pdsWork/AskHelp.py:48) 14/10/25
> 05:02:42 INFO DAGScheduler: Parents of final stage: List(Stage 1) 14/10/25
> 05:02:42 INFO DAGScheduler: Missing parents: List(Stage 1) 14/10/25 05:02:42
> INFO DAGScheduler: Submitting Stage 1 (PairwiseRDD[4] at RDD at
> PythonRDD.scala:261), which has no missing parents 14/10/25 05:02:42 INFO
> MemoryStore: ensureFreeSpace(9880) called with curMem=163705,
> maxMem=278302556 14/10/25 05:02:42 INFO MemoryStore: Block broadcast_1
> stored as values in memory (estimated size 9.6 KB, free 265.2 MB) 14/10/25
> 05:02:42 INFO DAGScheduler: Submitting 1 missing tasks from Stage 1
> (PairwiseRDD[4] at RDD at PythonRDD.scala:261) 14/10/25 05:02:42 INFO
> TaskSchedulerImpl: Adding task set 1.0 with 1 tasks 14/10/25 05:02:42 INFO
> TaskSetManager: Starting task 0.0 in stage 1.0 (TID 0, localhost,
> PROCESS_LOCAL, 1245 bytes) 14/10/25 05:02:42 INFO Executor: Running task 0.0
> in stage 1.0 (TID 0) 14/10/25 05:02:42 INFO Executor: Fetching http
> ://192.168.0.15:55111/files/AskHelp.py with timestamp 1414227761547 14/10/25
> 05:02:42 INFO Utils: Fetching http ://192.168.0.15:55111/files/AskHelp.py to
> /tmp/fetchFileTemp2903133416780112069.tmp 14/10/25 05:02:42 INFO
> CacheManager: Partition rdd_2_0 not found, computing it 14/10/25 05:02:42
> INFO HadoopRDD: Input split:
> file:/home/sid/Downloads/spark/pdsWork/input.txt:0+147355 14/10/25 05:02:42
> INFO deprecation: mapred.tip.id is deprecated. Instead, use
> mapreduce.task.id 14/10/25 05:02:42 INFO deprecation: mapred.task.id is
> deprecated. Instead, use mapreduce.task.attempt.id 14/10/25 05:02:42 INFO
> deprecation: mapred.task.is.map is deprecated. Instead, use
> mapreduce.task.ismap 14/10/25 05:02:42 INFO deprecation:
> mapred.task.partition is deprecated. Instead, use mapreduce.task.partition
> 14/10/25 05:02:42 INFO deprecation: mapred.job.id is deprecated. Instead,
> use mapreduce.job.id 14/10/25 05:02:42 ERROR PythonRDD: Python worker exited
> unexpectedly (crashed) org.apache.spark.api.python.PythonException:
> Traceback (most recent call last): File
> "/home/sid/Downloads/spark/python/pyspark/worker.py", line 75, in main
> command = pickleSer._read_with_length(infile) File
> "/home/sid/Downloads/spark/python/pyspark/serializers.py", line 146, in
> _read_with_length length = read_int(stream) File
> "/home/sid/Downloads/spark/python/pyspark/serializers.py", line 464, in
> read_int raise EOFError EOFError
>
>
>
>
>
> --
> View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/Python-code-crashing-on-ReduceByKey-if-I-return-custom-class-object-tp17317.html
> Sent from the Apache Spark User List mailing list archive at Nabble.com.
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
> For additional commands, e-mail: user-help@spark.apache.org
>
---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
For additional commands, e-mail: user-help@spark.apache.org