5

iPython Notebook を介して、ローカル マシンで Apache Spark の PySpark を使用してコードのプロトタイプを作成しています。正常に動作するように見えるコードをいくつか書きましたが、簡単な変更を加えると壊れてしまいます。

以下の最初のコード ブロックが機能します。2 番目のブロックは、指定されたエラーで失敗します。どんな助けにも本当に感謝します。このエラーは、Python オブジェクトのシリアル化に関係していると思われます。エラーは、TestClass を Pickle できないことを示しています。クラスをピクル可能にする方法に関する情報が見つかりません。ドキュメントには、「一般に、オブジェクトのすべての属性をピクルできる場合は、任意のオブジェクトをピクルできます。クラス、関数、およびメソッドをピクルすることはできません。オブジェクトをピクルすると、オブジェクトのクラスはピクルされず、何を識別する文字列にすぎません。これは、ほとんどのピクルスで問題なく機能します (ただし、ピクルスの長期保存に関する議論に注意してください)。TestClass を datetime クラスに置き換えてみたところ、問題なく動作しているように見えるため、これがわかりません。

とにかく、コード：

# ----------- 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

['テスト: こんにちは', 'テスト: こんにちは', 'テスト: こんにちは']

...終わり

しかし、「（1）この行を見る」とマークされた行（最後のマップ行）を削除して再実行すると、代わりに次のエラーが発生します。長いので、出力を投稿する前に、ここで質問を締めくくります。繰り返しますが、これについて助けていただければ幸いです。

前もって感謝します！

# ----------- 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 {0}{1}{2}.\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)*
4

2 に答える 2

2

コードの本体ではなく、独自のモジュールでクラスを定義する必要があることがわかりました。それを行ってからモジュールをインポートすると、pickle はオブジェクトを正常にピクルおよびアンピクルできます。その後、クラスは期待どおりに Spark で動作します。

于 2015-05-04T09:46:30.887 に答える
0

また、モジュールのクラス出力を定義します (私は pyspark で IPYTHON Notebook を実行しています)。

私は試した

  1. モジュールをspark/binにコピー - 失敗
  2. PythonVirtual Env へのコピー - 失敗
  3. Python 内の PATH の変更 - 失敗
  4. 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開発に役立つことを願っています

于 2015-07-03T08:19:03.253 に答える