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

Question

我已经设置了我的第一个spark集群（1个master,2个Worker）和一个iPython笔记本服务器,我已经设置了它来访问该集群。我正在运行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（最近一次调用last）in（）3 return 1 if xx+yy<1 else 0 4---->5 count=sc.parallelize（xrange（0,20））.map（sample）.reduce（lambda a,b：a+b）6 print“pi is大致%f”%（4.0*count/20）

/opt/spark-1.2.0/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返回reduce（f,vals）

/opt/spark-1.2.0/python/pyspark/rdd.pyc在collect（self）674“”“675中使用SCCallSiteSync（self.context）作为CSS:-->676 bytesInJava=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 in call（self,*args)536应答=self.gateway_client.send_command（command）537 return_value=get_return_value（应答,self.gateway_client,-->538 self.target_id,self.name)539 540对于temp_args中的temp_arg:

/opt/spark-1.2.0/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py in get_return_value（应答,gateway_client,target_id,name）298 Rise Py4JJavaError（299'调用{0}{1}{2}时出错.\n'.-->300格式(target_id,'.',name）,value）301 else：302 Rise Py4JError（

PY4JJavaError：调用O28.Collect时出错。：org.apache.spark.sparkException：作业由于阶段失败而中止：阶段0.0中的任务31失败了4次，最近一次失败：阶段0.0中丢失的任务31.3（TID 72,192.168.1.21):org.apache.spark.api.python.pythonException：追溯（最近一次调用）：文件“/opt/spark-1.2.0/python/pyspark/worker.py”，第107行，在main process（）文件“/opt/spark-1.2.0/python/pyspark/worker.py”，第98行，在process Serializer.dump_stream(func(split_index,iterator）,outfile）文件“/opt/spark-1.2.0/python dump_stream vs=list(itertools.islice（迭代器，批处理））文件”/opt/spark-1.2.0/python/pyspark/rdd.py“，第710行，在func initial=next（迭代器）文件”“中，第2行，在示例类型错误：”module“对象不可调用

在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$$anon$1.(pythonrdd.scala:174)在org.apache.spark.api.python.pythonrd.compute(pythonrdd.scala:96)在org.apache.spark.rdd.rdd.computeorreadcheckpoint(rdd.scala:263)在resulttask.runtask(resultask.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.1145)在java.util.concurrent.threadpoolexecutor$worker.run(threadpoolexecutor.615)在java.lang.thread.run（thread.java：745）

驱动程序stacktrace：在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)在AyBuffer.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)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.receiveMessage（actorcell.scala：516）在akka.actor.actorcell.invoke（actorcell.scala：487）在akka.dispatch.mailbox.processmailbox（mailbox.scala：238 KJointAsk.exec（AbstractDispatcher.Scala：393）在Scala.concurrent.ForkJoin.ForkJoinAsk.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()，这里是文档。