开发者社区

文档建议反馈控制台

最新优惠活动

文章/答案/技术大牛

发布

使用KafkaUtils.createDirectStream在Kafka中存储消息偏移量

Kafka是一个分布式流处理平台，用于高吞吐量、低延迟的数据传输和存储。Kafka中的消息以topic为单位进行组织和存储，每个topic可以有多个分区，每个分区可以有多个副本。

KafkaUtils.createDirectStream是Spark Streaming中用于从Kafka中读取数据的方法。它可以直接从Kafka的分区中读取数据，并且可以管理消息的偏移量。使用该方法可以实现实时处理Kafka中的消息。

使用KafkaUtils.createDirectStream在Kafka中存储消息偏移量的步骤如下：

创建一个Spark Streaming的上下文对象，指定批处理的时间间隔。
创建一个Kafka参数的Map对象，包含Kafka集群的地址、消费者组ID等信息。
创建一个Set对象，包含要从Kafka中读取数据的topic。
调用KafkaUtils.createDirectStream方法，传入上下文对象、Kafka参数、topic等参数，创建一个DStream对象。
在DStream上应用相应的转换操作，对读取到的数据进行处理。
启动Spark Streaming上下文对象。
等待数据的输入和处理。

使用KafkaUtils.createDirectStream的优势是可以直接从Kafka的分区中读取数据，而不需要通过Zookeeper来管理偏移量。这样可以简化系统的架构，并且提高了数据的可靠性和容错性。

KafkaUtils.createDirectStream的应用场景包括实时日志分析、实时数据处理、实时推荐系统等。在这些场景下，Kafka作为数据的中间件，可以实现高吞吐量、低延迟的数据传输和存储，而Spark Streaming可以对这些数据进行实时处理和分析。

腾讯云提供了一系列与Kafka相关的产品和服务，包括消息队列 CKafka、流计算 TDSQL-C、云原生消息队列 CMQ 等。您可以通过访问腾讯云官网了解更多详情和产品介绍：

相关搜索:Kafka连接消费者引用偏移量并存储在消息中在Flink中手动提交Kafka偏移量在Kafka中添加消息类型？如何使用KStreams -kafka和kafka-stream在spring Bean中记录偏移量使用Spark在Kafka上发布消息无法在kafka中永久保存消息在达到最新偏移量之前，不会将Kafka消息写入拼图文件如何使用有偏移量的服务器端的kafka消息？在Landoop kafka UI中哪里可以找到kafka消息头？消息密钥在Kafka流中的长度在Netcore和Kafka中每次使用一条消息使用springboot在KafkaConsumer中反序列化kafka消息如何使用danielwegener Kafka附加器定制Kafka消息中的ZonedDateatime 在python中手动提交kafka Direct Stream的偏移量 Kafka Consumer无法在偏移量提交后读取所有消息(error=OFFSET_OUT_OF_RANGE kafka消费者组在kafka-node中获取重复消息消息在kafka分区中是如何分布的？Python Kafka使用者在消息到达时不读取消息如何从Apache Nifi中上次提交的偏移量读取consumer中的Kafka消息？在几次尝试处理失败后将偏移量提交到reactor-kafka中的Kafka

相关搜索:

页面内容是否对你有帮助？

有帮助

没帮助

相关·内容

Spark Streaming 与 Kafka0.8 整合

在这里我们解释如何配置 Spark Streaming 以接收来自 Kafka 的数据。有两种方法，一种为使用 Receivers 和 Kafka 高级API的旧方法，以及不使用 Receivers 的新方法（在 Spark 1.3 中引入）。它们具有不同的编程模型，性能特征和语义保证。就目前的 Spark 版本而言，这两种方法都被为稳定的API。

02

如何管理Spark Streaming消费Kafka的偏移量（一）

最近工作有点忙，所以更新文章频率低了点，在这里给大家说声抱歉，前面已经写过在spark streaming中管理offset，但当时只知道怎么用，并不是很了解为何要那样用，最近一段时间又抽空看了一个github开源程序自己管理offset的源码，基本已经理解透彻了，当然这里面还包含了由于理解不透彻导致升级失败的一个案例，这个在下篇文章会分享出来。本篇我们先从理论的角度聊聊在Spark Streaming集成Kafka时的offset状态如何管理。 spark streaming 版本 2.1 kafka 版

07

一文告诉你SparkStreaming如何整合Kafka!

关于SparkStreaming从理论到实战的部分，博主已经在前面的博客中介绍了。本篇博客，为大家带来的是SparkStreaming整合Kafka的教程!

01

Spark Streaming 快速入门系列(4) | 一文告诉你SparkStreaming如何整合Kafka!

注意：读数据只能从Leader读，写数据也只能往Leader写，Follower会从Leader那里同步数据过来做副本！！！

02

2021年大数据Spark（四十三）：SparkStreaming整合Kafka 0.10 开发使用

The Spark Streaming integration for Kafka 0.10 is similar in design to the 0.8 Direct Stream approach；

02

Spark 中 Kafka Offset 管理

Spark在spark-streaming-kafka-0-10的API中实现了对Kafka Offset提交的API，在Spark消费过消息之后，提交消费过消息的Offset到Kafka里面，在Spark重启后，可以继续消费没有消费的消息，实现Exactly once的语义。

01

【Spark Streaming】Spark Streaming的使用

Spark Streaming是一个基于Spark Core之上的实时计算框架，可以从很多数据源消费数据并对数据进行实时的处理，具有高吞吐量和容错能力强等特点。

02

KafKa 代码实现

1.消费者 import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.serialization.StringDeserializer; import java.u

03

必读：Spark与kafka010整合

SparkStreaming与kafka010整合读本文之前，请先阅读之前文章：必读：再讲Spark与kafka 0.8.2.1+整合 Spark Streaming与kafka 0.10的整合，和0.8版本的direct Stream方式很像。Kafka的分区和spark的分区是一一对应的，可以获取offsets和元数据。API使用起来没有显著的区别。这个整合版本标记为experimental，所以API有可能改变。工程依赖首先，添加依赖。 groupId = org.apache.spark

07

Spark Streaming 快速入门系列(3) | DStream中如何创建数据源

测试过程中，可以通过使用ssc.queueStream(queueOfRDDs)来创建DStream，每一个推送到这个队列中的RDD，都会作为一个DStream处理。

02

SparkStreaming_Kafka_Redis整合

1.将kafka streaming 和 redis整合实现词频统计 Producer.class 生成数据daokafka package day14; /** * 创建一个生产者生成随机的key 和字母 * 用于实现实时流统计词频并存储到redis */ import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord;

03

Spark Streaming快速入门系列（7）

一般的大型集群和平台, 都需要对其进行监控的需求。要针对各种数据库, 包括 MySQL, HBase 等进行监控要针对应用进行监控, 例如 Tomcat, Nginx, Node.js 等要针对硬件的一些指标进行监控, 例如 CPU, 内存, 磁盘等

03

Spark Streaming 与 Kafka 整合的改进

Apache Kafka 正在迅速成为最受欢迎的开源流处理平台之一。我们在 Spark Streaming 中也看到了同样的趋势。因此，在 Apache Spark 1.3 中，我们专注于对 Spark Streaming 与 Kafka 集成进行重大改进。主要增加如下：

02

kafka系列-DirectStream

spark读取kafka数据流提供了两种方式createDstream和createDirectStream。两者区别如下： 1、KafkaUtils.createDstream 构造函数为KafkaUtils.createDstream(ssc, [zk], [consumer group id], [per-topic,partitions] ) 使用了receivers来接收数据，利用的是Kafka高层次的消费者api，对于所有的receivers接收到的数据将会保存在Spark executors中，然后通过Spark Streaming启动job来处理这些数据，默认会丢失，可启用WAL日志，该日志存储在HDFS上 A、创建一个receiver来对kafka进行定时拉取数据，ssc的rdd分区和kafka的topic分区不是一个概念，故如果增加特定主体分区数仅仅是增加一个receiver中消费topic的线程数，并不增加spark的并行处理数据数量 B、对于不同的group和topic可以使用多个receivers创建不同的DStream C、如果启用了WAL，需要设置存储级别，即KafkaUtils.createStream(….,StorageLevel.MEMORY_AND_DISK_SER) 2.KafkaUtils.createDirectStream 区别Receiver接收数据，这种方式定期地从kafka的topic+partition中查询最新的偏移量，再根据偏移量范围在每个batch里面处理数据，使用的是kafka的简单消费者api 优点: A、简化并行，不需要多个kafka输入流，该方法将会创建和kafka分区一样的rdd个数，而且会从kafka并行读取。 B、高效，这种方式并不需要WAL，WAL模式需要对数据复制两次，第一次是被kafka复制，另一次是写到wal中

02

SparkStreaming和Kafka基于Direct Approach如何管理offset

在之前的文章《解析SparkStreaming和Kafka集成的两种方式》中已详细介绍SparkStreaming和Kafka集成主要有Receiver based Approach和Direct Approach。同时对比了二者的优劣势，以及针对不同的Spark、Kafka集成版本处理方式的支持：

01

Spark Streaming的优化之路——从Receiver到Direct模式

随着大数据的快速发展，业务场景越来越复杂，离线式的批处理框架MapReduce已经不能满足业务，大量的场景需要实时的数据处理结果来进行分析、决策。Spark Streaming是一种分布式的大数据实时计算框架，他提供了动态的，高吞吐量的，可容错的流式数据处理，不仅可以实现用户行为分析，还能在金融、舆情分析、网络监控等方面发挥作用。个推开发者服务——消息推送“应景推送”正是应用了Spark Streaming技术，基于大数据分析人群属性，同时利用LBS地理围栏技术，实时触发精准消息推送，实现用户的精细化运营。此外，个推在应用Spark Streaming做实时处理kafka数据时，采用Direct模式代替Receiver模式的手段，实现了资源优化和程序稳定性提升。

04

Spark Streaming 整合 Kafka

Spark 针对 Kafka 的不同版本，提供了两套整合方案：spark-streaming-kafka-0-8 和 spark-streaming-kafka-0-10，其主要区别如下：

01

Spark Streaming的优化之路——从Receiver到Direct模式

随着大数据的快速发展，业务场景越来越复杂，离线式的批处理框架MapReduce已经不能满足业务，大量的场景需要实时的数据处理结果来进行分析、决策。Spark Streaming是一种分布式的大数据实时计算框架，他提供了动态的，高吞吐量的，可容错的流式数据处理，不仅可以实现用户行为分析，还能在金融、舆情分析、网络监控等方面发挥作用。个推开发者服务——消息推送“应景推送”正是应用了Spark Streaming技术，基于大数据分析人群属性，同时利用LBS地理围栏技术，实时触发精准消息推送，实现用户的精细化运营。此外，个推在应用Spark Streaming做实时处理kafka数据时，采用Direct模式代替Receiver模式的手段，实现了资源优化和程序稳定性提升。

02

必读：再讲Spark与kafka 0.8.2.1+整合

Kafka在0.8和0.10版本引入了新的消费者API，所以spark Streaming与kafka的整合提供了两个包。请根据你的集群选用正确的包。注意， 0.8和后期的版本0.9及0.10是兼

07

Spark Tips3: 在Spark Streaming job中读取Kafka messages及其offsetRange

在Spark Streaming job中读取Kafka topic(s)中的messages时，有时我们会需要同步记录下每次读取的messages的offsetRange。要达到这一目的，下面这两段代码（代码1和代码2）都是正确的，而且是等价的。代码1（正确）： ----------------------- JavaPairInputDStream<String, String> messages = KafkaUtils.createDirectStream( jssc,

❤️Spark的关键技术回顾，持续更新！【推荐收藏加关注】❤️

集群环境：CDH版本是5.14.0这个版本但由于spark对应的5.14.0的CDH版本的软件默认的版本是1.6.0同时阉割了SarkSQL，需要重新编译原因：因为Cloudera公司认为有了impala就不需要再使用sparkSQL的功能了，同时也是为了推广impala，所以直接阉割掉了sparkSQL的模块。解决：使用Apache的版本的spark来进行重新编译

02

基于大数据技术的开源在线教育项目三

本文代码可在开源项目https://github.com/SoundHearer/kuaiban中找到。

01

【Spark Streaming】Spark Day11：Spark Streaming 学习笔记

将每批次数据状态，按照Key与以前状态，使用定义函数【updateFunc】进行更新，示意图如下：

01

Spark综合性练习(Spark，Kafka，Spark Streaming，MySQL)

之前刚学Spark时分享过一篇磨炼基础的练习题，➤Ta来了，Ta来了，Spark基础能力测试题Ta来了!，收到的反馈还是不错的。于是，在正式结课Spark之后，博主又为大家倾情奉献一道关于Spark的综合练习题，希望大家能有所收获✍

01

Kafuka面试（整合Kafka两种模式区别）

开发中我们经常会利用SparkStreaming实时地读取kafka中的数据然后进行处理，在spark1.3版本后，kafkaUtils里面提供了两种创建DStream的方法:

02

Spark Streaming + Kakfa 编程指北

本文简述如何结合 Spark Streaming 和 Kakfa 来做实时计算。截止目前（2016-03-27）有两种方式：

04

Spark Tips4: Kafka的Consumer Group及其在Spark Streaming中的“异动”(更新)

按照Kafka官方的说法（http://kafka.apache.org/08/introduction.html），某一特定topic对于相同group id的clients采用queuing机制，也就是说topic中的每个message只能被多个group id相同的consumer instance(process或者machine)中的一个读取一次。使用Kafka的High Level Consumer API (kafka.javaapi.consumer.ConsumerConnector 的

Spark Streaming Join「建议收藏」

三种思路各有优劣，使用时注意一下。这里总结在计算引擎Spark Streaming上做Join。

02

Spark流式状态管理

通常使用Spark的流式框架如Spark Streaming，做无状态的流式计算是非常方便的，仅需处理每个批次时间间隔内的数据即可，不需要关注之前的数据，这是建立在业务需求对批次之间的数据没有联系的基础之上的。

02

Spark Streaming + Kafka整合

参考官网 http://spark.apache.org/docs/2.1.0/streaming-kafka-0-8-integration.html 之前先确保以下操作： 1、先启动ZK：./zkServer.sh start 2、启动Kafka：./kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties 3、创建topic： ./kafka-topics.sh --create --zookeeper hadoo

05

Spark Streaming 中使用 zookeeper 保存 offset 并重用 Java版

最近中使用spark Streaming +kafka,由于涉及到金额，所以需要保证at only one, 而网上关于java版的kafka offset回写zk的资料少之又少，于是总结一下，希望可以为广大使用java的友友们提供参考!这里采用的是Direct Approach的方式.

02

Spark Streaming——Spark第一代实时计算引擎

虽然SparkStreaming已经停止更新，Spark的重点也放到了 Structured Streaming ，但由于Spark版本过低或者其他技术选型问题，可能还是会选择SparkStreaming。SparkStreaming对于时间窗口，事件时间虽然支撑较少，但还是可以满足部分的实时计算场景的，SparkStreaming资料较多，这里也做一个简单介绍。

01

spark连接kafka工具类

版权声明：本文为博主原创，欢迎转载，转载请标明出处 Blog Address:http://blog.csdn.net/jsjsjs1789 https://blog.csdn.net/jsjsjs1789/article/details/82226508

01

基于Direct的方式

这种新的不基于Receiver的直接方式，是在Spark 1.3中引入的，从而能够确保更加健壮的机制。替代掉使用Receiver来接收数据后，这种方式会周期性地查询Kafka，来获得每个topic+partition的最新的offset，从而定义每个batch的offset的范围。当处理数据的job启动时，就会使用Kafka的简单consumer api来获取Kafka指定offset范围的数据。

02

Kafka与Spark Streaming整合

Spark Streaming是一个可扩展，高吞吐，容错能力强的实时流式处理处理系统。一般的系统架构图是，数据从一个源点，经过Sparing Streaming处理，最后汇聚到一个系统。Spark Streaming的数据来源可以非常丰富，比如Kafka, Flume, Twitter, ZeroMQ, Kinesis 或者是任何的TCP sockets程序。对于数据的处理，Spark Streaming提供了非常丰富的高级api，例如map，redue，joini和窗口函数等等。数据处理完成后，可以存储到其他地方，比如文件系统，对象存储，数据库。典型的数据处理流程图：

07

Spark Streaming 项目实战 (2) | 从 Kafka中消费数据

在依赖选择上spark-streaming-kafka此次选用0-10_2.11而非0-08_2.11

01

Spark Stream对接kafka 源码分析

本文会讲解Spark Stream是如何与Kafka进行对接的，包括DirectInputStream和KafkaRDD是如何与KafkaConsumer交互的

02

spark streaming消费指定的topic和partition并手动更新offset

直接上代码 scala版的 import kafka.common.TopicAndPartition import kafka.message.MessageAndMetadata import kafka.serializer.Decoder import org.apache.spark.SparkException import org.apache.spark.rdd.RDD import org.apache.spark.streaming.StreamingContext impo

02

（3）sparkstreaming从kafka接入实时数据流最终实现数据可视化展示

（1）sparkstreaming从kafka接入实时数据流最终实现数据可视化展示，我们先看下整体方案架构：

04

Spark2.3.1+Kafka0.9使用Direct模式消费信息异常

在验证kafka属性时不能使用scala默认的类，需要指定kafka带的类 createDirectStream[String, String, StringDecoder, StringDecoder]其中StringDecoder必须是kafka.serializer.StringDecoder

06

Spark Streaming——Spark第一代实时计算引擎

虽然SparkStreaming已经停止更新，Spark的重点也放到了 Structured Streaming ，但由于Spark版本过低或者其他技术选型问题，可能还是会选择SparkStreaming。 SparkStreaming对于时间窗口，事件时间虽然支撑较少，但还是可以满足部分的实时计算场景的，SparkStreaming资料较多，这里也做一个简单介绍。

01

学习笔记:StructuredStreaming入门（十二）

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-6wtQxLP6-1626354186973)(/img/image-20210506154426999.png)]

01

Note_Spark_Day12： StructuredStreaming入门

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-6wtQxLP6-1626354186973)(/img/image-20210506154426999.png)]

01

pyspark streaming简介和消费 kafka示例

/spark/examples/src/main/python/streaming

02

Spark Streaming消费Kafka数据的两种方案

下午的时候翻微信看到大家在讨论Spark消费Kafka的方式，官网中就有答案，只不过是英文的，当然很多博客也都做了介绍，正好我的收藏夹中有一篇文章供大家参考。文章写的通俗易懂，搭配代码，供大家参考。

04

Spark Streaming应用与实战全攻略

有一块业务主要是做爬虫抓取与数据输出，通过大数据这边提供的SOA服务入库到HBase,架构大致如下：

03

来看看一个大二学生的Spark练习题

23. 总成绩大于150分，且数学大于等于70，且年龄大于等于19岁的学生的平均成绩是多少？

06

kafka主题offset各种需求修改方法

简要：开发中，常常因为需要我们要认为修改消费者实例对kafka某个主题消费的偏移量。具体如何修改？为什么可行？其实很容易，有时候只要我们换一种方式思考，如果我自己实现kafka消费者，我该如何让我们的消费者代码如何控制对某一个主题消费，以及我们该如何实现不同消费者组可以消费同一个主题的同一条消息，一个消费组下不同消费者消费同一个主题的不同消息。如果让你实现该框架该如何实现？

01

揭开Spark Streaming神秘面纱⑥ - Spark Streaming结合 Kafka 两种不同的数据接收方式比较

DirectKafkaInputDStream 只在 driver 端接收数据，所以继承了 InputDStream，是没有 receivers 的

01

Spark Kafka 基于Direct自己管理offset

在Spark Streaming中，目前官方推荐的方式是createDirectStream方式，但是这种方式就需要我们自己去管理offset。目前的资料大部分是通过scala来实现的，并且实现套路都是一样的，我自己根据scala的实现改成了Java的方式，后面又相应的实现。 Direct Approach 更符合Spark的思维。我们知道，RDD的概念是一个不变的，分区的数据集合。我们将kafka数据源包裹成了一个KafkaRDD，RDD里的partition 对应的数据源为kafka的partition。唯一的区别是数据在Kafka里而不是事先被放到Spark内存里。其实包括FileInputStream里也是把每个文件映射成一个RDD。

02

扫码

添加站长进交流群

领取专属 10元无门槛券

手把手带您无忧上云

扫码加入开发者社群

相关资讯

热门标签

活动推荐

运营活动

活动名称

广告关闭