python3+spark2.1+kafka0.8+sparkStreaming

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了python3+spark2.1+kafka0.8+sparkStreaming相关的知识,希望对你有一定的参考价值。

python代码:

import time
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
from operator import add


sc = SparkContext(master="local[1]",appName="PythonSparkStreamingRokidDtSnCount")
ssc = StreamingContext(sc, 2)
zkQuorum = localhost:2181
topic = {rokid:1}
groupid = "test-consumer-group"
lines = KafkaUtils.createStream(ssc, zkQuorum, groupid, topic)
lines1 = lines.flatMap(lambda x: x.split("\n"))
valuestr = lines1.map(lambda x: x.value.decode())
valuedict = valuestr.map(lambda x:eval(x))
message = valuedict.map(lambda x: x["message"])
rdd2 = message.map(lambda x: (time.strftime("%Y-%m-%d",time.localtime(float(x.split("\u0001")[0].split("\u0002")[1])/1000))+"|"+x.split("\u0001")[1].split("\u0002")[1],1)).map(lambda x: (x[0],x[1]))
rdd3 = rdd2.reduceByKey(add)
rdd3.saveAsTextFiles("/tmp/wordcount")
rdd3.pprint()
ssc.start()
ssc.awaitTermination()

执行SparkStreaming:

spark/bin/spark-submit --jars spark-streaming-kafka-0-8-assembly_2.11-2.1.0.jar ReadFromKafkaStreaming.py

 

其中spark-streaming-kafka-0.98-assembly_2.11-2.1.0.jar从以下网站下载
http://search.maven.org

 

作为入门参考。


以上是关于python3+spark2.1+kafka0.8+sparkStreaming的主要内容,如果未能解决你的问题,请参考以下文章

kafka spark streaming例子——TODO 没有成功

Kafka0.8.2官方文档中文版系列-topic配置参数

Kafka0.8.2.1删除topic逻辑

Kafka0.8.2官方文档中文版系列-API

Kafka0.8.2官方文档中文版系列-入门指南

Kafka0.8性能测试报告