sparkstreaming监听hdfs目录_大数据系列之Spark Streaming接入Kafka数据

Spark Streaming官方提供Receiver-based和Direct Approach两种方法接入Kafka数据，本文简单介绍两种方式的pyspark实现。

1、Spark Streaming接入Kafka方式介绍

Spark Streaming 官方提供了两种方式读取Kafka数据：

  1. 一是Receiver-based Approach。该种读取模式官方最先支持，并在Spark 1.2提供了数据零丢失(zero-data loss)的支持；

  2. 一是Direct Approach (No Receivers)。该种读取方式在Spark 1.3引入。

1.1 Receiver-based Approach

Receiver-based的Kafka读取方式是基于Kafka高阶(high-level) api来实现对Kafka数据的消费。在提交Spark Streaming任务后，Spark集群会划出指定的Receivers来专门、持续不断、异步读取Kafka的数据，读取时间间隔以及每次读取offsets范围可以由参数来配置。读取的数据保存在Receiver中，具体StorageLevel方式由用户指定，诸如MEMORY_ONLY等。当driver 触发batch任务的时候，Receivers中的数据会转移到剩余的Executors中去执行。在执行完之后，Receivers会相应更新ZooKeeper的offsets。如要确保at least once的读取方式，可以设置spark.streaming.receiver.writeAheadLog.enable为true。具体Receiver执行流程见下图：

67702ffd747ead37eae665f8fec06fb1.png

  1. 需要借助Write Ahead Logs 来保证数据的不丢失，如果启用了Write Ahead Logs复制到文件系统如HDFS，那么storage level需要设置成StorageLevel.MEMORY_AND_DISK_SER，也就是KafkaUtils.createStream(…, StorageLevel.MEMORY_AND_DISK_SER)

  2. 在Receiver的方式中，Spark中的partition和kafka中的partition并不是相关的，所以如果我们加大每个topic的partition数量，仅仅是增加线程来处理由单一Receiver消费的主题。但是这并没有增加Spark在处理数据上的并行度

  3. 对于不同的Group和topic我们可以使用多个Receiver创建不同的Dstream来并行接收数据，之后可以利用union来统一成一个Dstream

1.2 Direct Approach (No Receivers)

Direct方式采用Kafka简单的consumer api方式来读取数据，无需经由ZooKeeper，此种方式不再需要专门Receiver来持续不断读取数据。当batch任务触发时，由Executor读取数据，并参与到其他Executor的数据计算过程中去。由drive来决定读取多少offsets，并将offsets交由checkpoints来维护。将触发下次batch任务，再由Executor读取Kafka数据并计算。从此过程可以发现Direct方式无需Receiver读取数据，而是需要计算时再读取数据，所以Direct方式的数据消费对内存的要求不高，只需要考虑批量计算所需要的内存即可；另外batch任务堆积时，也不会影响数据堆积。其具体读取方式如下图：

3e4b0a03f1eb6a04b9ffb9a634b48a17.png

  1. 简化的并行：在Receiver的方式中提到创建多个Receiver之后利用union来合并成一个Dstream的方式提高数据传输并行度。而在Direct方式中，Kafka中的partition与RDD中的partition是一一对应的并行读取Kafka数据，这种映射关系也更利于理解和优化。

  2. 高效：在Receiver的方式中，为了达到0数据丢失需要将数据存入Write Ahead Log中，这样在Kafka和日志中就保存了两份数据，浪费！而第二种方式不存在这个问题，只要我们Kafka的数据保留时间足够长，我们都能够从Kafka进行数据恢复。

  3. 精确一次：在Receiver的方式中，使用的是Kafka的高阶API接口从Zookeeper中获取offset值，这也是传统的从Kafka中读取数据的方式，但由于Spark Streaming消费的数据和Zookeeper中记录的offset不同步，这种方式偶尔会造成数据重复消费。而第二种方式，直接使用了简单的低阶Kafka API，Offsets则利用Spark Streaming的checkpoints进行记录，消除了这种不一致性。

2、Spark Streaming接入Kafka数据实现

以wordcount统计为例，kafka生产端输入词组，Spark端读取kafka流数据，并统计词频

2.1 Receiver方式收取数据

1)Import KafkaUtils并创建DStream

from pyspark.streaming.kafka import KafkaUtils

kafkaStream = KafkaUtils.createStream(streamingContext, \
     [ZK quorum], [consumer group id], [per-topic number of Kafka partitions to consume])

  1. ZK Quorum：Zookeeper quorum (hostname:port,hostname:port,..)

  2. Groupid：消费者的groupid

  3. Topics：{topic_name : numPartitions}

2)具体实现代码如下：

from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils

if __name__ == "__main__":
    #if len(sys.argv) != 3:
    #    print("Usage: kafka_wordcount.py ", file=sys.stderr)
    #    exit(-1)

    sc = SparkContext(appName="PythonStreamingKafkaWordCount")
    ssc = StreamingContext(sc, 10)

    zkQuorum = "192.168.112.101:2181,192.168.112.102:2181,192.168.112.103:2181"
    groupid = "spark-streaming-consumer"
    topic = {"kafka_spark_test1":0,"kafka_spark_test1":1,"kafka_spark_test1":2}
    #zkQuorum, topic = sys.argv[1:]
    kvs = KafkaUtils.createStream(ssc, zkQuorum, groupid, topic)
    lines = kvs.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()

    ssc.start()
    ssc.awaitTermination()

在Spark目录执行命令：

spark-submit --master yarn --deploy-mode client --jars jars/spark-streaming-kafka-0-8-assembly_2.11-2.3.1.jar /usr/local/spark/ipynotebook/03-kafka2streaming-01.py
2.2 Direct方式收取数据

1)Import KafkaUtils并创建DStream

from pyspark.streaming.kafka import KafkaUtils
 directKafkaStream = KafkaUtils.createDirectStream(ssc, [topic], {"metadata.broker.list": brokers})

  • ssc：StreamingContext

  • topics：消费的topics清单

  • {"metadata.broker.list": brokers}：kafka参数，可以指定为 metadata.broker.list或bootstrap.servers

  • 默认情况下，从每个kafka分区的最新的offset进行消费，如果在kafka参数中设置了auto.offset.reset 为smallest，则会从最小的offset进行消费

  • 如果希望保存每个批量消费的kafka offset，可以进行如下操作：

offsetRanges = []

def storeOffsetRanges(rdd):
global offsetRanges
     offsetRanges = rdd.offsetRanges()
return rdd

def printOffsetRanges(rdd):
for o in offsetRanges:
print "%s %s %s %s" % (o.topic, o.partition, o.fromOffset, o.untilOffset)

 directKafkaStream \
     .transform(storeOffsetRanges) \
     .foreachRDD(printOffsetRanges)

如果希望使用基于Zookeeper的Kafka监控，也可以通过这种方法展现Streaming的进程。

2)具体实现代码如下：

from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils

offsetRanges = []

def storeOffsetRanges(rdd):
    global offsetRanges
    offsetRanges = rdd.offsetRanges()
    return rdd

def printOffsetRanges(rdd):
    for o in offsetRanges:
        print("%s %s %s %s" % (o.topic, o.partition, o.fromOffset, o.untilOffset))

if __name__ == "__main__":
    #if len(sys.argv) != 3:
    #    print("Usage: direct_kafka_wordcount.py ", file=sys.stderr)
    #    exit(-1)

    sc = SparkContext(appName="PythonStreamingDirectKafkaWordCount")
    ssc = StreamingContext(sc, 10)

    #brokers, topic = sys.argv[1:]
    topic="kafka_spark_test1"
    brokers = "192.168.112.101:9092,192.168.112.102:9092,192.168.112.103:9092"
    kvs = KafkaUtils.createDirectStream(ssc, [topic], {"metadata.broker.list": brokers})
    lines = kvs.map(lambda x: x[1])
    counts = lines.flatMap(lambda line: line.split(" ")) \
        .map(lambda word: (word, 1)) \
        .reduceByKey(lambda a, b: a+b)
    kvs.transform(storeOffsetRanges).foreachRDD(printOffsetRanges)
    counts.pprint()

    ssc.start()                   # Start the computation
    ssc.awaitTermination()        # Wait for the computation to terminate

在Spark根目录执行命令：

spark-submit --master yarn --deploy-mode client --jars jars/spark-streaming-kafka-0-8-assembly_2.11-2.3.1.jar /usr/local/spark/ipynotebook/03-kafka2streaming-02.py
2.3 Kafka生产者配置

Kafka集群环境的安装配置，参考之前的文档"大数据系列之Kafka集群环境部署"中相关内容

1)启动zookeeper

[root@tango-centos01 kafka_2.11-1.1.0]# nohup ./bin/zookeeper-server-start.sh  ./config/zookeeper.properties &
[root@tango-centos02 kafka_2.11-1.1.0]# nohup ./bin/zookeeper-server-start.sh  ./config/zookeeper.properties &
[root@tango-centos03 kafka_2.11-1.1.0]# nohup ./bin/zookeeper-server-start.sh  ./config/zookeeper.properties &

2)启动Kafka集群

[root@tango-centos01 kafka_2.11-1.1.0]# nohup ./bin/kafka-server-start.sh  ./config/server.properties &
[root@tango-centos02 kafka_2.11-1.1.0]# nohup ./bin/kafka-server-start.sh  ./config/server.properties &
[root@tango-centos03 kafka_2.11-1.1.0]# nohup ./bin/kafka-server-start.sh  ./config/server.properties &

3)创建Kafka topic

[root@tango-centos01 kafka_2.11-1.1.0]# ./bin/kafka-topics.sh --create --zookeeper 192.168.112.101:2181,192.168.112.102:2181,192.168.112.103:2181 --replication-factor 2 --partitions 3 --topic kafka_spark_test1
Created topic "kafka_spark_test1".

创建名为kafka_spark_test1 的Topic，复制因子设为2，同时分区数为3，注意，分区数是read parallelisms的最大值

4)查看Topic详情

[root@tango-centos01 kafka_2.11-1.1.0]# ./bin/kafka-topics.sh --describe --zookeeper 192.168.112.101:2181,192.168.112.102:2181,192.168.112.103:2181 --topic kafka_spark_test1
Topic:kafka_spark_test1 PartitionCount:3        ReplicationFactor:2     Configs:
        Topic: kafka_spark_test1     Partition: 0    Leader: 2    Replicas: 2,3   Isr: 2,3
        Topic: kafka_spark_test1     Partition: 1    Leader: 3    Replicas: 3,1   Isr: 3,1
        Topic: kafka_spark_test1     Partition: 2    Leader: 1    Replicas: 1,2   Isr: 1,2

指定--zookeeper选项的值为192.168.112.101:2181,192.168.112.102:2181,192.168.112.103:2181，对应的Topic，即刚创建的kafka_spark_test1

2.4 Kafka-Spark Streaming流测试

1)下载依赖的jars包

31a90c07093d8e3ed63275c3d4b0f8e0.png

2)启动kafka生产者

[root@tango-centos01 kafka_2.11-1.1.0]# ./bin/kafka-console-producer.sh --broker-list 192.168.112.101:9092 --topic kafka_spark_test1

3)运行Spark Streaming流数据处理程序

[root@tango-spark01 spark-2.3.0]# spark-submit --master yarn --deploy-mode client --jars jars/spark-streaming-kafka-0-8-assembly_2.11-2.3.1.jar /usr/local/spark/ipynotebook/03-kafka2streaming-01.py
[root@tango-spark01 spark-2.3.0]# spark-submit --master yarn --deploy-mode client --jars jars/spark-streaming-kafka-0-8-assembly_2.11-2.3.1.jar /usr/local/spark/ipynotebook/03-kafka2streaming-02.py

4)在Kafka生产端输入流数据

[root@tango-centos01 kafka_2.11-1.1.0]# ./bin/kafka-console-producer.sh --broker-list 192.168.112.101:9092 --topic kafka_spark_test1
>hello world
>hello tango hello
>hello tango tango

5)终端打印结果

-------------------------------------------
Time: 2018-08-08 11:03:15
-------------------------------------------
(u'tango', 2)
(u'hello', 1)

465db79ecec5ab184f83ef034ac4a93b.png

6)登录SparkWeb UI，查看Spark Streaming的的运行情况

a) spark-submit时候指定spark-submit --master spark://192.168.112.121:7077才能在8080端口看到数据

5204816405c5738434957e90283af29c.png

f19c5eb3b72cd5122c779ecb6c60a51e.png

b) 如果通过yarn模式调度，可通过8088端口查看

745de24154a094ea190a012eedd97859.png

2.5 Spark写入Kafka

1)安装Kafka插件

Pyspark访问Kafka需要使用到kafka安装包，使用以下命令安装：

pip install --no-index --find-links=../kafka-1.3.5-py2.py3-none.any.whl kafka

2)调用KafkaProducer模块，spark作为生产者将数据传输到kafka端

from kafka import KafkaProducer

to_kafka = KafkaProducer(bootstrap_servers=broker_list)
to_kafka.send(topic_name,send_msg,encode(‘utf8’))
to_kafka.flush()

参考资料

  1. http://spark.apache.org/docs/latest/streaming-kafka-integration.html

  2. 大数据系列之Kafka集群环境部署

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/365316.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

php职业认证,如何用 PHP 进行 HTTP 认证

php职业认证,如何用 PHP 进行 HTTP 认证

PHP 的 HTTP 认证机制在 PHP 以 Apache 模块方式运行时才有效，因此该功能不适用于 CGI 版本。在 Apache 模块的 PHP 脚本中，可以用 header() 函数来向客户端浏览器发送“Authentication Required”信息，使其弹出一个用户名/密码输入窗口。当用…
阅读更多...
时间服务器

时间服务器

时间服务器配置： 1 安装软件包: [rootlocalhost ~]# yum install ntp –y 2 修改配置文件 [rootlocalhost ~]# vim /etc/ntp.conf # 允许内网其他机器同步时间 192.168.1.0该网段 restrict 192.168.1.0 mask 255.255.255.0 nomodify notrap 允许任何ip的客户机都可以…
阅读更多...
使用 HTML5 Canvas 绘制出惊艳的水滴效果

使用 HTML5 Canvas 绘制出惊艳的水滴效果

HTML5 在不久前正式成为推荐标准，标志着全新的 Web 时代已经来临。在众多 HTML5 特性中，Canvas 元素用于在网页上绘制图形，该元素标签强大之处在于可以直接在 HTML 上进行图形操作，具有极大的应用价值。 这里分享一个惊艳的 Canva…
阅读更多...
Java EE 8：社区想要什么？

Java EE 8：社区想要什么？

在进行了之前的两次社区调查之后，第一次调查于2013年12月发布，第二次调查于2014年1月发布，Oracle呼吁企业社区对Java EE 8进行第三次也是最后一次调查 。该调查的截止日期为2014年3月24日，12日。太平洋时间。 值得注意的是&#…
阅读更多...
mysql 字符串类型 char varchar

mysql 字符串类型 char varchar

字符类型用在存储名字、邮箱地址、家庭住址等描述性数据 char指的是定长字符，varchar指的是变长字符 #官网：https://dev.mysql.com/doc/refman/5.7/en/char.html #注意：char和varchar括号内的参数指的都是字符的长度#char类型：定长…
阅读更多...
二叉树遍历规则

二叉树遍历规则

树的遍历顺序大体分为三种：前序遍历（先根遍历、先序遍历），中序遍历（中根遍历），后序遍历（后根遍历）。 如图所示二叉树： 前序遍历：前序遍历可以记为…
阅读更多...
vue vue的table表格自适应_vue table autoHeight(vue 表格自动高度)

vue vue的table表格自适应_vue table autoHeight(vue 表格自动高度)

小编主要做的都是后台管理系统，采用布局多为头部、左侧菜单栏，右侧内容，头部和菜单栏固定位置，内容部分如果很长就会出现滚动条(iview和element都提供了布局容器)，后台管理系统多为表格的增删改查，so&#…
阅读更多...
php网页的注册界面设计,HTML开发博客之注册页面设计（一）

php网页的注册界面设计,HTML开发博客之注册页面设计（一）

CSS文件的引入新建文件reg.html文件首先我们来分析网页布局这是我们页面完成后的效果，网页分为三部分头部，主体，和底部我们按照这个顺序开始编写。头部导航栏的编写html>用户注册页面首页科技资讯心情随笔资源收藏图文图片留言板登陆/注册…
阅读更多...
Arctext.js - 基于 CSS3 jQuery 的文本弯曲效果

Arctext.js - 基于 CSS3 jQuery 的文本弯曲效果

Arctext.js 是基于 Lettering.js 的文本旋转插件，根据设置的旋转半径准确计算每个字母的旋转弧度并均匀分布。虽然 CSS3 也能够实现字符旋转效果，但是要让安排每个字母都沿着弯曲路径排布相当的复杂，结合 Arctext.js 则可以轻松实现这个效果。…
阅读更多...
在JDT中使用Java 8 Lambda

在JDT中使用Java 8 Lambda

旧 Curmudgeon 认识Smalltalk的Dude 在修改Eclipse Java开发工具 （JDT）项目正在开发的Java 8支持时，我一直在使用这种语言。 我承认我对Java 8中的lambda有点不满意。 当然，这来自于知道Smalltalk （和LISP / Schem…
阅读更多...
[POJ1463] Strategic game

[POJ1463] Strategic game

题目链接： 传送门 题目大意： Bob非常享受玩电脑游戏的过程，尤其是策略游戏，但是在有些时候，他因为不能在第一时间找到最佳的策略而十分伤心。 现在，他遇到了一个问题。他必须保卫一个中世纪的城市&#xff…
阅读更多...
zabbix邮件告警

zabbix邮件告警

zabbix邮件告警调用第三方邮件服务来发送邮件。 [roottiandong ~]# yum install mailx -y 修改配置文件 [roottiandong ~]# vim /etc/mail.rc 67 set from15600857257163.com smtpsmtp.163.com 68 set smtp-auth-user15600857257163.com 69 set smtp-auth-password密码&#xf…
阅读更多...
模糊推理机制 matlab,基于模糊综合评判推理机制的学生素质评价系统设计与实现...

模糊推理机制 matlab,基于模糊综合评判推理机制的学生素质评价系统设计与实现...

第22卷第2期2002年2月文章编号：(2002)l00l-908l2-008l-03计算机应用ComputerAppIicationsVoI.22,No.2Feb.,2002基于模糊综合评判推理机制的学生素质评价系统设计与实现尹世群(西南师范大学计算机与信息科学学院，重庆4007l5)摘要：根据模糊数学…
阅读更多...
SweetAlert – 替代 Alert 的漂亮的提示效果

SweetAlert – 替代 Alert 的漂亮的提示效果

Sweet Alert 是一个替代传统的 JavaScript Alert 的漂亮提示效果。SweetAlert 自动居中对齐在页面中央，不管您使用的是台式电脑，手机或平板电脑看起来效果都很棒。另外提供了丰富的自定义配置选择，可以灵活控制。 在线演示 插件下载 您可…
阅读更多...
onpagefinished等了很久才执行_今天自律了吗？停课不停锻炼 才是战疫正确姿势

onpagefinished等了很久才执行_今天自律了吗？停课不停锻炼 才是战疫正确姿势

近日，中青校媒面向全国915名高校学生发起关于“宅家运动”情况的调查，发现15.39%被调查者在家期间会严格执行锻炼计划，39.96%选择间歇性完成制订的运动目标，还有44.65%在家很少运动。(3月18日《中国青年报》)新冠肺炎疫情发生&…
阅读更多...
Spring Data Couchbase 1.0 GA发布

Spring Data Couchbase 1.0 GA发布

Spring Data Couchbase 1.0 GA版本已发布！ 该项目是Spring Data项目的一部分，该项目旨在为新数据存储提供熟悉且一致的基于Spring的编程模型，同时保留特定于存储的功能。 Spring Data Couchbase项目提供了与Couchbase Server数据库的集成。 …
阅读更多...
网络验证php接口逆向,一个专利查错的逆向（网络验证）

网络验证php接口逆向,一个专利查错的逆向（网络验证）

[C#] 纯文本查看 复制代码bool flag true;if (!fileInfo.CreationTime.ToString().Equals(configData.fileCreateTime))//这里是校验License，直接删除掉{MessageBox.Show("认证文件已损坏！");flag false;Application.Exit();}if (flag){Lice…
阅读更多...
kickstart-G

kickstart-G

感觉自己很蠢&#xff0c;large数据只能交一次&#xff0c;忘记这回事了 A题 O(n^2)解法&#xff0c;用vector<set> 缓存j后面的数据&#xff0c;减少一重循环 1 #include <string>2 #include <vector>3 #include<iostream>4 #include<cstdio>5 …
阅读更多...
Ocrad.js – JS 实现 OCR 光学字符识别

Ocrad.js – JS 实现 OCR 光学字符识别

Ocrad.js 相当于是 Ocrad 项目的纯 JavaScript 版本&#xff0c;使用 Emscripten 自动转换。这是一个简单的 OCR &#xff08;光学字符识别&#xff09;程序&#xff0c;可以扫描图像中的文字回文本。 不像 GOCR.js&#xff0c;Ocrad.js 被设计成一个端口&#xff0c;而不是围绕…
阅读更多...
c++ ptree判断是否存在节点_CTreeCtrl 查找 等于某值 的节点，并设置为高亮

c++ ptree判断是否存在节点_CTreeCtrl 查找 等于某值 的节点，并设置为高亮

查找节点&#xff0c;并返回该节点HTREEITEM GroupTree::FindTreeItem(HTREEITEM item,CString& strText){HTREEITEM hFind;if(item NULL)return NULL;while(item!NULL){if(m_pTree->GetItemText(item) strText)return item;if(m_pTree->ItemHasChildren(item))…
阅读更多...
最新文章

Copyright @ 2022~2023