星火直接流與卡夫卡

-1

我想實現的卡夫卡星火DirectStream下面是代碼星火直接流與卡夫卡

con = SparkConf().setMaster("local[2]").setAppName("Streamer") 
sc = SparkContext(conf=con) 
ssc = StreamingContext(sc, 10) 
ssc.checkpoint("checkpoint") 
#consumer(ssc) 
kstream = KafkaUtils.createDirectStream(ssc, kafkaParams = {"bootstrap_servers": 'localhost:9092'},topics = ['twitter']) 
tweets = kstream.map(lambda line: line.decode('ascii')) 
text = tweets.map(lambda line: line.split(',')) 
print text.collect()

當我運行它，我得到以下錯誤。哪裏不對？

Traceback (most recent call last): 
    File "/Users/vinayaka/Desktop/BigData/HW3/SentimentAnalyser/consumer.py", line 20, in <module> 
    main() 
    File "/Users/vinayaka/Desktop/BigData/HW3/SentimentAnalyser/consumer.py", line 14, in main 
    kstream = KafkaUtils.createDirectStream(ssc, kafkaParams = {"bootstrap_servers": 'localhost:9092'},topics = ['twitter']) 
    File "/usr/local/Cellar/apache-spark/2.1.0/libexec/python/lib/pyspark.zip/pyspark/streaming/kafka.py", line 130, in createDirectStream 
    File "/usr/local/Cellar/apache-spark/2.1.0/libexec/python/lib/py4j-0.10.4-src.zip/py4j/java_gateway.py", line 1133, in __call__ 
    File "/usr/local/Cellar/apache-spark/2.1.0/libexec/python/lib/py4j-0.10.4-src.zip/py4j/protocol.py", line 319, in get_return_value 
py4j.protocol.Py4JJavaError: An error occurred while calling o23.createDirectStreamWithoutMessageHandler. 
: org.apache.spark.SparkException: Must specify metadata.broker.list or bootstrap.servers 
    at org.apache.spark.streaming.kafka.KafkaCluster$SimpleConsumerConfig$$anonfun$9.apply(KafkaCluster.scala:417) 
    at org.apache.spark.streaming.kafka.KafkaCluster$SimpleConsumerConfig$$anonfun$9.apply(KafkaCluster.scala:417) 
    at scala.Option.getOrElse(Option.scala:121) 
    at org.apache.spark.streaming.kafka.KafkaCluster$SimpleConsumerConfig$.apply(KafkaCluster.scala:417) 
    at org.apache.spark.streaming.kafka.KafkaCluster.config(KafkaCluster.scala:53) 
    at org.apache.spark.streaming.kafka.KafkaCluster.getPartitionMetadata(KafkaCluster.scala:130) 
    at org.apache.spark.streaming.kafka.KafkaCluster.getPartitions(KafkaCluster.scala:119) 
    at org.apache.spark.streaming.kafka.KafkaUtils$.getFromOffsets(KafkaUtils.scala:211) 
    at org.apache.spark.streaming.kafka.KafkaUtilsPythonHelper.createDirectStream(KafkaUtils.scala:720) 
    at org.apache.spark.streaming.kafka.KafkaUtilsPythonHelper.createDirectStreamWithoutMessageHandler(KafkaUtils.scala:688) 
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 
    at java.lang.reflect.Method.invoke(Method.java:498) 
    at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244) 
    at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357) 
    at py4j.Gateway.invoke(Gateway.java:280) 
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132) 
    at py4j.commands.CallCommand.execute(CallCommand.java:79) 
    at py4j.GatewayConnection.run(GatewayConnection.java:214) 
    at java.lang.Thread.run(Thread.java:745)

來源

2017-04-11 user3482434

你確定它是bootstrap_servers？我覺得應該是bootstrap.servers

kstream = KafkaUtils.createDirectStream(ssc, ['twitter'], {"bootstrap.servers": 'localhost:9092'})

你可以看看here瞭解更多詳情。確保你去python選項卡。

來源

2017-04-11 05:11:06 radbrawler

我試過kafkaParams = {「bootstrap.servers」：'localhost：9092'}和kafkaParams = {「metadata.broker.list」：'localhost：9092'}。仍然得到相同的錯誤。 – user3482434

你試過'kstream = KafkaUtils.createDirectStream（ssc，['twitter']，{「bootstrap.servers」：'localhost：9092'}）'？我知道他們是相同的，但以防萬一 – radbrawler

星火直接流與卡夫卡

回答

相關問題