Kafka + Spark Stream实时WordCount

栏目: 编程工具 · 发布时间: 5年前

内容简介:Word Count简直就是大数据届的hello world。所谓Word Count就是计算一行或者一段文本中英文单词的出现个数(英文单词以空格分隔)。这篇文章示范了如何使用Kafka + Spark Streaming来实现一个实时版本的Word Count。这个范例比较简单,仅仅有助于跑通流程。在实时运算时,一个很重要的问题就是:时间窗。比如说,统计实时的在线人数,当有新用户上线时,在线人数+1,但是过15分钟后,如果该用户的“最后活跃时间”仍是上线时间,那么此时就要去除它。Word Count这个

Word Count简直就是大数据届的hello world。所谓Word Count就是计算一行或者一段文本中英文单词的出现个数(英文单词以空格分隔)。这篇文章示范了如何使用Kafka + Spark Streaming来实现一个实时版本的Word Count。这个范例比较简单,仅仅有助于跑通流程。在实时运算时,一个很重要的问题就是:时间窗。比如说,统计实时的在线人数,当有新用户上线时,在线人数+1,但是过15分钟后,如果该用户的“最后活跃时间”仍是上线时间,那么此时就要去除它。

Word Count这个例子没有时间窗的概念,所以有点过于简单,但对于初次接触的同学,理解实时计算是什么样的还是有一点帮助吧。

# coding=utf-8
# 提交Spark作业
# $SPARK_HOME/bin/spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.1 /data/pyjobs/test/kafka-wordcount.py

from __future__ import print_function
from pyspark.sql import SparkSession
from pyspark.sql.functions import explode
from pyspark.sql.functions import split
from pyspark.sql import functions as F

bootstrapServers = "kafka1:9092,kafka2:9092,kafka3:9092"

spark = SparkSession\
    .builder\
    .appName("StructuredKafkaWordCount")\
    .getOrCreate()

# 基于来自kafka的数据流,创建dataframe
lines = spark\
    .readStream\
    .format("kafka")\
    .option("kafka.bootstrap.servers", bootstrapServers)\
    .option("subscribe", "test.wordcount.input")\
    .option("failOnDataLoss", False)\
    .option("group.id", "wordcount-group3")\
    .load()\
    .selectExpr("CAST(value AS STRING)")

# 将单行数据拆分,转成多行数据
words = lines.select(
    explode(split(lines.value, ' ')).alias('word')
)

# 对单词进行分组,并计算总数
wordCounts = words.groupBy('word').count()

# 将两列数据合并成单列数据
wordCounts = wordCounts.select(F.concat(F.col("word"), F.lit("|"), F.col("count").cast("string")).alias("value"))

# 测试时,可以不将结果写入kafka,直接输出到控制台
# query = wordCounts \
#     .writeStream \
#     .outputMode("complete") \
#     .format("console") \
#     .start()

# 将结果输出到 test.wordcount.output
query = wordCounts \
    .writeStream \
    .format('kafka') \
    .outputMode('update') \
    .option("kafka.bootstrap.servers", bootstrapServers) \
    .option('checkpointLocation', '/spark/job-checkpoint') \
    .option("topic", "test.wordcount.output") \
    .start()

query.awaitTermination()

提交Spark作业之前,需要先创建两个Kafka的topic:test.wordcount.input,用于录入数据,由Spark读取,进行运算后,再写入到 test.wordcount.output 中:

# 创建和写入 test.wordcount.input
# bin/kafka-topics.sh --zookeeper zookeeper1:2181/kafka --create --topic test.wordcount.input --replication-factor 2 --partitions 6
# bin/kafka-console-producer.sh --broker-list kafka1:9092 --topic test.wordcount.input

# 创建和读取 test.wordcount.output
# bin/kafka-topics.sh --zookeeper zookeeper1:2181/kafka --create --topic test.wordcount.output --replication-factor 2 --partitions 6
# bin/kafka-console-consumer.sh --bootstrap-server kafka1:9092 --topic test.wordcount.output

关于kafka的控制台命令,可以参看: Kafka分布式消息系统(通过控制台访问) - Part.4

有一点需要注意的:在执行spark-submit的时候,需要加上--packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.1 选项,因为要读取/写入Kafka topic。

Kafka + Spark Stream实时WordCount
Spark Stream实时运算WordCount

上图左边,是Spark作业的提交窗口;右上方是字符的录入窗口,右下方是结果窗口;每当在右上方输入句子时,便会在右下方实时计算出单词的出现数量。

此时如果想进一步处理(例如进行显示),只需要编写一个kafka的客户端,从test.wordcount.output中读取数据就可以了。

至此,就完成了实时Word Count这个范例。以后会再做一个加入“时间窗”的更贴近实际项目的范例吧。

感谢阅读,希望这篇文章能给你带来帮助!

Kafka + Spark Stream实时WordCount


以上就是本文的全部内容,希望本文的内容对大家的学习或者工作能带来一定的帮助,也希望大家多多支持 码农网

查看所有标签

猜你喜欢:

本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们

Beginning XML with DOM and Ajax

Beginning XML with DOM and Ajax

Sas Jacobs / Apress / 2006-06-05 / USD 39.99

Don't waste time on 1,000-page tomes full of syntax; this book is all you need to get ahead in XML development. Renowned web developer Sas Jacobs presents an essential guide to XML. Beginning XML with......一起来看看 《Beginning XML with DOM and Ajax》 这本书的介绍吧!

HTML 编码/解码
HTML 编码/解码

HTML 编码/解码

UNIX 时间戳转换
UNIX 时间戳转换

UNIX 时间戳转换

HEX CMYK 转换工具
HEX CMYK 转换工具

HEX CMYK 互转工具