mit Kafka Direkt Streaming Funken

Question

ich versuche

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() 

Wenn ich es laufen bekomme ich die folgenden Fehler den Code unten Funkendirect für Kafka zu implementieren. Was ist falsch?

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) 

Answer 1

Sind Sie sicher, dass es bootstrap_servers ist? Ich denke, es sollte bootstrap.servers

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

Sie sich here, um weitere Informationen zu suchen. Stellen Sie sicher, dass Sie auf die Python-Registerkarte gehen.