Spark 流式传输（运行示例与 spark-submit）答案

【问题标题】：Spark streaming (run-example vs spark-submit)Spark 流式传输（运行示例与 spark-submit）
【发布时间】：2015-11-08 08:13:58
【问题描述】：

这是Spark with Flume (configuration/classpath?)的后续问题

在尝试了几件事后将问题存储起来，我现在遇到的问题是

$spark-submit --jars /opt/scala/spark-streaming-flume_2.10-1.5.1.jar --master local[*] /home/user/spark/FlumeStreaming.py

from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.flume import FlumeUtils

sc = SparkContext(appName="Newapp")
strm = StreamingContext(sc,1)

flume = FlumeUtils.createStream(strm,"localhost",9999)

lines = flume.map(lambda x: x[1])
counts = lines.flatMap(lambda line: line.split(" ")) \
        .map(lambda word: (word, 1)) \
        .reduceByKey(lambda a, b: a+b)
counts.pprint()

strm.start()
strm.awaitTermination()

15/11/07 23:55:09 WARN TaskSetManager: Lost task 0.0 in stage 0.0 (TID 0, localhost): java.lang.NoClassDefFoundError: org/apache/flume/source/avro/AvroSourceProtocol
        at org.apache.spark.streaming.flume.FlumeReceiver.responder$lzycompute(FlumeInputDStream.scala:147)
        at org.apache.spark.streaming.flume.FlumeReceiver.responder(FlumeInputDStream.scala:146)
        at org.apache.spark.streaming.flume.FlumeReceiver.initServer(FlumeInputDStream.scala:163)
        at org.apache.spark.streaming.flume.FlumeReceiver.onStart(FlumeInputDStream.scala:170)
        at org.apache.spark.streaming.receiver.ReceiverSupervisor.startReceiver(ReceiverSupervisor.scala:148)
        at org.apache.spark.streaming.receiver.ReceiverSupervisor.start(ReceiverSupervisor.scala:130)
        at org.apache.spark.streaming.scheduler.ReceiverTracker$ReceiverTrackerEndpoint$$anonfun$9.apply(ReceiverTracker.scala:542)
        at org.apache.spark.streaming.scheduler.ReceiverTracker$ReceiverTrackerEndpoint$$anonfun$9.apply(ReceiverTracker.scala:532)
        at org.apache.spark.SparkContext$$anonfun$37.apply(SparkContext.scala:1984)
        at org.apache.spark.SparkContext$$anonfun$37.apply(SparkContext.scala:1984)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
        at org.apache.spark.scheduler.Task.run(Task.scala:88)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
        at java.lang.Thread.run(Thread.java:744)
Caused by: java.lang.ClassNotFoundException: org.apache.flume.source.avro.AvroSourceProtocol
        at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
        at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
        at java.security.AccessController.doPrivileged(Native Method)
        at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
        ... 16 more

15/11/07 23:55:09 ERROR TaskSetManager: Task 0 in stage 0.0 failed 1 times; aborting job
15/11/07 23:55:09 ERROR ReceiverTracker: Deregistered receiver for stream 0: Stopped by driver

我的问题是，它与 spark 示例中提供的 flume_wordcount.py 代码相同，示例中的代码可以正常工作，但我的版本不起作用。不同之处在于它使用 run-example 运行另一种使用 spark-submit 运行的方式，这指向了类路径和 jar 文件的管理方式。有什么我应该做的吗？

【问题讨论】：

标签： python spark-streaming flume-ng

【解决方案1】：

解决了这个问题，必须有正确的 jar 并将其传递给 spark-submit

$spark-submit --jars /path/to/spark-streaming-flume-assembly*.jar FlumeStreaming.py localhost 12345

【讨论】：

【解决方案2】：

您不应使用具有依赖关系的 spark-streaming-flume_2.10-1.5.1.jar。您可以在http://mvnrepository.com/artifact/org.apache.spark/spark-streaming-flume-assembly_2.10 中下载程序集 jar，然后像上面那样运行。

希望对你有帮助。

【讨论】：