当 RDD 包含用户定义的类时,为啥 Apache PySpark top() 会失败?
Posted
技术标签:
【中文标题】当 RDD 包含用户定义的类时,为啥 Apache PySpark top() 会失败?【英文标题】:Why does Apache PySpark top() fail when the RDD contains a user defined class?当 RDD 包含用户定义的类时,为什么 Apache PySpark top() 会失败? 【发布时间】:2015-03-18 19:18:57 【问题描述】:我正在通过 iPython Notebook 在本地机器上使用 Apache Spark 的 PySpark 对一些代码进行原型设计。我编写了一些似乎可以正常工作的代码,但是当我对其进行简单更改时,它就会中断。
下面的第一个代码块有效。第二个块因给定错误而失败。非常感谢任何帮助。我怀疑这个错误与序列化 Python 对象有关。错误说它不能 Pickle TestClass。我找不到有关如何使我的班级可以泡菜的信息。文档说“通常,如果您可以腌制该对象的每个属性,则可以腌制任何对象。类,函数和方法不能腌制-如果腌制一个对象,则该对象的类不会腌制,只是一个标识什么的字符串它所属的类。这对大多数泡菜都有效(但请注意关于泡菜长期储存的讨论)。“。我不明白这一点,因为我尝试用 datetime 类替换我的 TestClass 并且一切似乎都很好。
不管怎样,代码:
# ----------- This code works -----------------------------
class TestClass(object):
def __init__(self):
self.teststr = 'Hello'
def __str__(self):
return self.teststr
def __repr__(self):
return self.teststr
def test(self):
return 'test: 0'.format(self.teststr)
#load multiple text files into list of RDDs, concatenate them, then remove headers
trip_rdd = trip_rdds[0]
for rdd in trip_rdds[1:]:
trip_rdd = trip_rdd.union(rdd)
#filter out header rows from result
trip_rdd = trip_rdd.filter(lambda r: r != header)
#split the line, then convert each element to a dictionary
trip_rdd = trip_rdd.map(lambda r: r.split(','))
trip_rdd = trip_rdd.map(lambda r, k = header_keys: dict(zip(k, r)))
trip_rdd = trip_rdd.map(convert_trip_dict)
#trip_rdd = trip_rdd.map(lambda d, ps = g_nyproj_str: Trip(d, ps))
#originally I map the given dictionaries to a 'Trip' class I defined with various bells and whistles.
#I've simplified to using TestClass above and still seem to get the same error
trip_rdd = trip_rdd.map(lambda t: TestClass())
trip_rdd = trip_rdd.map(lambda t: t.test()) #(1) Watch this row
print trip_rdd.count()
temp = trip_rdd.top(3)
print temp
print '...done'
以上代码返回如下:
347098
['test: Hello', 'test: Hello', 'test: Hello']
...完成
但是,当我删除标记为“(1) watch this row”的行(最后一个地图行)并重新运行时,我得到了以下错误。它很长,所以在发布输出之前,我将在这里结束我的问题。再次,我非常感谢您对此的帮助。
提前致谢!
# ----------- This code FAILS -----------------------------
class TestClass(object):
def __init__(self):
self.teststr = 'Hello'
def __str__(self):
return self.teststr
def __repr__(self):
return self.teststr
def test(self):
return 'test: 0'.format(self.teststr)
#load multiple text files into list of RDDs, concatenate them, then remove headers
trip_rdds = [sc.textFile(f) for f in trip_files]
trip_rdd = trip_rdds[0]
for rdd in trip_rdds[1:]:
trip_rdd = trip_rdd.union(rdd)
#filter out header rows from result
trip_rdd = trip_rdd.filter(lambda r: r != header)
#split the line, then convert each element to a dictionary
trip_rdd = trip_rdd.map(lambda r: r.split(','))
trip_rdd = trip_rdd.map(lambda r, k = header_keys: dict(zip(k, r)))
trip_rdd = trip_rdd.map(convert_trip_dict)
#trip_rdd = trip_rdd.map(lambda d, ps = g_nyproj_str: Trip(d, ps))
#originally I map the given dictionaries to a 'Trip' class I defined with various bells and whistles.
#I've simplified to using TestClass above and still seem to get the same error
trip_rdd = trip_rdd.map(lambda t: TestClass())
trip_rdd = trip_rdd.map(lambda t: t.test()) #(1) Watch this row
print trip_rdd.count()
temp = trip_rdd.top(3)
print temp
print '...done'
输出: 347098
*---------------------------------------------------------------------------
Py4JJavaError Traceback (most recent call last)
<ipython-input-76-6550318a5d5b> in <module>()
29 #count them
30 print trip_rdd.count()
---> 31 temp = trip_rdd.top(3)
32 print temp
33 print '...done'
C:\Programs\Apache\Spark\spark-1.2.0-bin-hadoop2.4\python\pyspark\rdd.pyc in top(self, num, key)
1043 return heapq.nlargest(num, a + b, key=key)
1044
-> 1045 return self.mapPartitions(topIterator).reduce(merge)
1046
1047 def takeOrdered(self, num, key=None):
C:\Programs\Apache\Spark\spark-1.2.0-bin-hadoop2.4\python\pyspark\rdd.pyc in reduce(self, f)
713 yield reduce(f, iterator, initial)
714
--> 715 vals = self.mapPartitions(func).collect()
716 if vals:
717 return reduce(f, vals)
C:\Programs\Apache\Spark\spark-1.2.0-bin-hadoop2.4\python\pyspark\rdd.pyc in collect(self)
674 """
675 with SCCallSiteSync(self.context) as css:
--> 676 bytesInJava = self._jrdd.collect().iterator()
677 return list(self._collect_iterator_through_file(bytesInJava))
678
C:\Programs\Coding\Languages\Python\Anaconda_32bit\Conda\lib\site-packages\py4j-0.8.2.1-py2.7.egg\py4j\java_gateway.pyc in __call__(self, *args)
536 answer = self.gateway_client.send_command(command)
537 return_value = get_return_value(answer, self.gateway_client,
--> 538 self.target_id, self.name)
539
540 for temp_arg in temp_args:
C:\Programs\Coding\Languages\Python\Anaconda_32bit\Conda\lib\site-packages\py4j-0.8.2.1-py2.7.egg\py4j\protocol.pyc in get_return_value(answer, gateway_client, target_id, name)
298 raise Py4JJavaError(
299 'An error occurred while calling 012.\n'.
--> 300 format(target_id, '.', name), value)
301 else:
302 raise Py4JError(
Py4JJavaError: An error occurred while calling o463.collect.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 1 in stage 49.0 failed 1 times, most recent failure: Lost task 1.0 in stage 49.0 (TID 99, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "C:\Programs\Apache\Spark\spark-1.2.0-bin-hadoop2.4\python\pyspark\worker.py", line 107, in main
process()
File "C:\Programs\Apache\Spark\spark-1.2.0-bin-hadoop2.4\python\pyspark\worker.py", line 98, in process
serializer.dump_stream(func(split_index, iterator), outfile)
File "C:\Programs\Apache\Spark\spark-1.2.0-bin-hadoop2.4\python\pyspark\serializers.py", line 231, in dump_stream
bytes = self.serializer.dumps(vs)
File "C:\Programs\Apache\Spark\spark-1.2.0-bin-hadoop2.4\python\pyspark\serializers.py", line 393, in dumps
return cPickle.dumps(obj, 2)
PicklingError: Can't pickle <class '__main__.TestClass'>: attribute lookup __main__.TestClass failed
at org.apache.spark.api.python.PythonRDD$$anon$1.read(PythonRDD.scala:137)
at org.apache.spark.api.python.PythonRDD$$anon$1.<init>(PythonRDD.scala:174)
at org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:96)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:230)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61)
at org.apache.spark.scheduler.Task.run(Task.scala:56)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196)
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)
Driver stacktrace:
at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1214)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1203)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1202)
at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1202)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:696)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:696)
at scala.Option.foreach(Option.scala:236)
at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:696)
at org.apache.spark.scheduler.DAGSchedulerEventProcessActor$$anonfun$receive$2.applyOrElse(DAGScheduler.scala:1420)
at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
at org.apache.spark.scheduler.DAGSchedulerEventProcessActor.aroundReceive(DAGScheduler.scala:1375)
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)*
【问题讨论】:
作者在这里 - 刚刚尝试将 TestClass 的实例腌制到文件 F,关闭 python,重新打开 python 并从 F 中解压缩,并得到错误"module object has no attribute 'TestClass'"
。
【参考方案1】:
原来你必须在自己的模块中定义你的类,而不是在代码的主体中。如果你这样做然后导入模块,pickle 能够成功地腌制和取消腌制对象。然后,该课程可以按照您的预期与 Spark 一起使用。
【讨论】:
【参考方案2】:我还要定义模块的类输出(我正在使用 pyspark 运行 IPYTHON Notebook)。
我试过了
-
将模块复制到 spark/bin - 失败
复制到 PythonVirtual Env - 失败
在 Python 中修改 PATH - 失败
编写 PIP 设置模块 - 并安装 PIP - 工作
代码示例
from army2 import army2
People=["1,Maj,123","2,Pvt,333","3,Col,999"]
rrd1=sc.parallelize(People)
rrd2=rrd1.map( lambda y: army2(y))
Army2 模块
我的高度复杂的军队模块看起来像这样......
class army2:
def __init__(self,list_of_data):
fields = list_of_data.split(",")
self.id =fields[0]
self.rank=fields[1]
self.num=fields[2]
def __repr__(self):
return "We're in the Army2 Now"
if __name__ == "__main__":
a=army2('1,"tim",345')
print a
Setup.py
我的 pip setup.py 看起来像这样
import os
from setuptools import setup
def read(*paths):
"""Build a file path from *paths* and return the contents."""
with open(os.path.join(*paths), 'r') as f:
return f.read()
setup(
name='army',
version='0.1.0',
description='Make a small army.',
long_description=(read('README.rst')),
url='http://github.com/timseed/junk/',
license='GNU',
author='Gen McArthur',
author_email='bounce@bouncy.nowhere',
py_modules=['army2'],
include_package_data=True,
classifiers=[
'Development Status :: 5 - Production/Stable',
'Intended Audience :: Developers',
'Natural Language :: English',
'Operating System :: OS Independent',
'Programming Language :: Python',
'Programming Language :: Python :: 2',
'Programming Language :: Python :: 2.6',
'Programming Language :: Python :: 2.7',
'Programming Language :: Python :: 3',
'Programming Language :: Python :: 3.3',
'Topic :: Software Development :: Libraries :: Python Modules',
],
)
我希望这有助于其他人的 pyspark 开发
【讨论】:
以上是关于当 RDD 包含用户定义的类时,为啥 Apache PySpark top() 会失败?的主要内容,如果未能解决你的问题,请参考以下文章
为啥当我没有为我的 openGLWidget 创建额外的类时 glRotate 只工作一次?