无法从 Python 运行 Apache Spark 的 Pi 示例

Question

我已经设置了我的第一个 spark 集群（1 个 master，2 个 worker）和一个 iPython notebook 服务器，我已经设置它来访问集群。我正在运行 Anaconda 的工作人员，以确保每个盒子上的 python 设置都是正确的。 iPy 笔记本服务器似乎已正确设置所有内容，并且我能够初始化 Spark 并将作业推送出去。但是，这项工作失败了，我不知道如何排除故障。代码如下：

from pyspark import SparkContext
from numpy import random
CLUSTER_URL = 'spark://192.168.1.20:7077'
sc = SparkContext( CLUSTER_URL, 'pyspark')
def sample(p):
    from numpy import random
    x, y = random(), random()
    return 1 if x*x + y*y < 1 else 0

count = sc.parallelize(xrange(0, 20)).map(sample).reduce(lambda a, b: a + b)
print "Pi is roughly %f" % (4.0 * count / 20)

这是错误：

Py4JJavaError Traceback（最近调用 最后）在（） 3 如果 xx + yy 5 count = sc.parallelize(xrange(0, 20)).map(sample).reduce(lambda a, b: a + b) 6 print "Pi 大约是 %f" % (4.0 * count / 20)

/opt/spark-1.2.0/python/pyspark/rdd.pyc in reduce(self, f) 第713章 714 --> 715 vals = self.mapPartitions(func).collect() 716 如果值： 第717章

/opt/spark-1.2.0/python/pyspark/rdd.pyc in collect(self) 第674章 675 与 SCCallSiteSync(self.context) 作为 css： --> 676 字节InJava = self._jrdd.collect().iterator() 677 返回列表（self._collect_iterator_through_file（bytesInJava）） 第678章

/opt/spark-1.2.0/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py 在 调用(self, *args) 第536章 第537章 --> 538 self.target_id, self.name) 539 temp_args 中的 temp_arg 为 540：

/opt/spark-1.2.0/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py get_return_value(answer, gateway_client, target_id, name) 第298章 299 '调用 {0}{1}{2} 时发生错误。\n'。 --> 300 格式（target_id，'.'，名称），值） 301 其他： 302 引发 Py4JError(

Py4JJavaError：调用 o28.collect 时出错。 ： org.apache.spark.SparkException：作业因阶段失败而中止： 阶段 0.0 中的任务 31 失败 4 次，最近一次失败：丢失任务 31.3 在阶段 0.0 (TID 72, 192.168.1.21): org.apache.spark.api.python.PythonException: Traceback (最近 最后调用）：文件“/opt/spark-1.2.0/python/pyspark/worker.py”，行 107、主要 process() 文件“/opt/spark-1.2.0/python/pyspark/worker.py”，第 98 行，正在处理中 serializer.dump_stream(func(split_index, iterator), outfile) 文件“/opt/spark-1.2.0/python/pyspark/serializers.py”，第 227 行，在 转储流 vs = list(itertools.islice(iterator, batch)) 文件“/opt/spark-1.2.0/python/pyspark/rdd.py”，第 710 行，在 func initial = next(iterator) File "", line 2, in sample TypeError: 'module' object is not callable

在 org.apache.spark.api.python.PythonRDD$$anon$1.read(PythonRDD.scala:137) 在 org.apache.spark.api.python.PythonRDD$$anon$1.(PythonRDD.scala:174) 在 org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:96) 在 org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263) 在 org.apache.spark.rdd.RDD.iterator(RDD.scala:230) 在 org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61) 在 org.apache.spark.scheduler.Task.run(Task.scala:56) 在 org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196) 在 java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) 在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) 在 java.lang.Thread.run(Thread.java:745)

驱动程序堆栈跟踪：在 org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1214) 在 org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1203) 在 org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1202) 在 scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) 在 scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47) 在 org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1202) 在 org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:696) 在 org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:696) 在 scala.Option.foreach(Option.scala:236) 在 org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:696) 在 org.apache.spark.scheduler.DAGSchedulerEventProcessActor$$anonfun$receive$2.applyOrElse(DAGScheduler.scala:1420) 在 akka.actor.Actor$class.aroundReceive(Actor.scala:465) 在 org.apache.spark.scheduler.DAGSchedulerEventProcessActor.aroundReceive(DAGScheduler.scala:1375) 在 akka.actor.ActorCell.receiveMessage(ActorCell.scala:516) 在 akka.actor.ActorCell.invoke(ActorCell.scala:487) 在 akka.dispatch.Mailbox.processMailbox(Mailbox.scala:238) 在 akka.dispatch.Mailbox.run(Mailbox.scala:220) 在 akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:393) 在 scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) 在 scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) 在 scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) 在 scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)

我什至不确定从哪里开始调试/诊断此问题，因此我们将不胜感激。如果有帮助，很高兴发布其他日志。

Answer 1

numpy.random是一个Python包，你不能用random()调用它。

我猜你想使用random.random()，这里是documentation。