来自kafka的spark流如何指定轮询事件的截止时间
？
在使用Spark Streaming处理来自Kafka的流数据时，可以通过设置参数来指定轮询事件的截止时间。这个截止时间决定了每个批次的数据处理时间窗口。
在Spark Streaming中，可以使用createDirectStream方法来创建一个直连的Kafka数据流。在创建数据流时，可以通过ConsumerStrategies类的assign方法来指定要消费的Kafka分区，并通过ConsumerConfig类的MAX_POLL_INTERVAL_MS_CONFIG参数来设置轮询事件的截止时间。
具体步骤如下：
- 导入相关的类和包：
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.spark.streaming.kafka010.ConsumerStrategies
import org.apache.spark.streaming.kafka010.KafkaUtils- 创建Spark Streaming上下文：
val sparkConf = new SparkConf().setAppName("KafkaSparkStreaming")
val ssc = new StreamingContext(sparkConf, Seconds(5))- 设置Kafka参数：
val kafkaParams = Map[String, Object](
ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "kafka-broker1:9092,kafka-broker2:9092",
ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
ConsumerConfig.GROUP_ID_CONFIG -> "group-id",
ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "latest",
ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG -> "60000" // 设置轮询事件的截止时间为60秒
)- 创建Kafka数据流：
val topics = Array("topic1", "topic2")
val stream = KafkaUtils.createDirectStream[String, String](
ssc,
PreferConsistent,
ConsumerStrategies.Assign[String, String](topics, kafkaParams)
)通过以上步骤，我们可以创建一个直连的Kafka数据流，并通过ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG参数设置轮询事件的截止时间为60秒。这样，Spark Streaming将会在每个60秒的时间窗口内处理来自Kafka的数据。
注意：以上示例中的参数和配置仅供参考，实际使用时需要根据具体情况进行调整。
推荐的腾讯云相关产品：腾讯云消息队列 CMQ、腾讯云云服务器 CVM、腾讯云云原生容器服务 TKE、腾讯云数据库 TencentDB、腾讯云对象存储 COS、腾讯云区块链服务 TBCS。你可以通过访问腾讯云官网了解更多关于这些产品的详细信息和使用指南。
相关·内容
3回答
我有一个spark流媒体应用程序，它在一天结束的时候运行并消费上游application.Currently发送的kafka事件，上游应用程序整天都在推送新数据，而我的消费者最终消费了它。我想根据截止时间限制消费的事件，比如6点daily.Is，有一种方法可以指定截止时间来限制消费的事件，比如卡夫卡事件
浏览 23提问于2019-01-12得票数 2
回答已采纳
我通读了spark structured streaming文档，我想知道spark structured是如何确定事件已经迟到的？它是否将事件时间与处理时间进行比较？1)这个处理时间是从哪里来的？由于它的流，它是否假设有人可能正在使用具有处理时间戳的
浏览 0提问于2018-02-26得票数 4
1回答
我有一个用例，在这个用例中，我需要阅读kafka中的主题中的事件，并在Spark中处理它们。我需要处理与同一元素相关的事件(事件消息中指定了元素标识符)，这些事件发生在一个时间框架内。如果相关事件分布在卡夫卡分区中，那么我们如何读取所有相关事件，给出spark的多执行者体系结构？
是否有可能强制所有相关
浏览 2提问于2020-06-29得票数 0
来自MySQL和Kafka主题的事件数据。我们将从MySQL加载历史数据，然后切换到Kafka，以便当前的data.Mysql ID也会流入Kafka。因此，消息包含事件ID。Kafka流事件将使用spark作业进行处理，并最终将数据摄取到报表中。 问题:如何加载历史数据并实时切换到Kafka源。 1-创建临时使用者组以存储所有分区的当前偏移量和
浏览 18提问于2020-01-23得票数 0
2回答
我正在开发一个实时流应用程序，该应用程序可以从Kafka broker中轮询数据，并且我正在调整以前默认使用Spark结构化流的代码(带有微批处理)。但是，我不知道如何使用连续流而不是微批量流来获得类似的行为。这是一段可以工作的代码： .foreachBatch(foreach_batch_func) \这就是我到目前为止对连续<em
浏览 1提问于2020-09-29得票数 5
2回答
我们正在开放源码Apache连接器之上与mongodb kafka connetor合作，以获取从Mongo到HDFS的json数据。我们有kafka使用者，它读取kafka中的数据更改并将它们写入hdfs文件。我们需要根据预定日期触发卡夫卡消息。
浏览 9提问于2022-09-23得票数 0
回答已采纳
1回答
基于Kafka的分布式调度器
、、、、
我们有一个使用案例，在Kafka中，我们每天都会收到数百万个事件。每个事件将如下所示：所以对于每个id，我们可以有多个事件。我们需要生成警报，以防在例如xx天之后，在这些事件流中没有接收到特定id的任何一个状态。id 123的例子，如果我们
浏览 15提问于2021-03-01得票数 0
1回答
我有一个要求，从多个卡夫卡topicsAvro的流媒体，并把他们在格林梅利与小修改的有效载荷。val query1 = df.writeStream.start()
sp
浏览 2提问于2020-11-06得票数 1
回答已采纳
1回答
火花流写卡夫卡与延迟后x分钟
、、、、
我们有一个火花流应用程序。体系结构如下Spark应用程序正在使用qubole/ Kinesis进行来自Kinesis的结构化流。然后将数据进行聚合，然后推送给卡夫卡。我们的用例需要延迟4分钟才能进入卡夫卡。 .writeStream .format
浏览 0提问于2019-06-06得票数 2
回答已采纳
在使用Spark Structured Streaming和Kinesis Streams时，我在重新处理流中积累的数据时遇到了读取不平衡的情况(而不是从最新数据读取)。下图显示了作为流的一部分的运动碎片的读取速度的差异。这使得spark作业丢弃了许多事件，因为事件时间非常不同的事件会被混淆，而那些被认为是旧
浏览 1提问于2021-07-04得票数 1
4回答
我正在做一个在数据流上运行机器学习算法的POC。Spark Streaming -->聚合多个表中的数据-->在数据流上运行MLLib -->产生输出。问题：
火花流和卡夫卡流有什么区别？我如何才能与KStreams +火花流+机器学习结合起来？我的想法是持续训练测试数据，而不是批量训练。
浏览 2提问于2016-12-14得票数 9
Kafka是很常见的。所以很多公司都在使用它。我完全理解Kafka和Spark是如何工作的，我对他们都很有经验。我不理解的是用例。为什么你要把Kafka和Spark一起使用，而不仅仅是Spark呢？在我看来，Kafka的主要用途是作为ETL管道中的中转区，用于实时(流)数据。
我假设有一个数据源集群，数据最初存储在其中。例如，它可以是Vertic
浏览 3提问于2019-06-17得票数 3
我使用星火结构流和连接两个来自卡夫卡主题的流。我注意到，流查询对于每条记录大约需要15秒。在下面的截图中，阶段id 2需要15s。为什么会这样？ .builder .getOrCreate
.readStream
.format(&quo
浏览 1提问于2018-11-27得票数 1
回答已采纳
Spark streaming应用程序实时接收来自大量IoT设备的数据。但它们都是少量的数据。整体流程看起来像是-> Iot -> Kafka(1个主题/所有数据) ->火花流(过滤错误日志) -> DB(保存) ->警报屏幕 有没有使用spark或python进行实时日志分析的好方法？
浏览 13提问于2020-11-19得票数 0
回答已采纳
我开始使用spark structured。我通过waterMark从kafka topic (startOffset: latest)获取readStream，按事件时间和窗口时长分组，并写入kafka topic。我的问题是，在spark结构化流媒体作业之前，我如何处理写入kafka主题的数据？
一开始我试着用‘`startOffset: with’来运行。但是kafka topic中的
浏览 87提问于2018-11-26得票数 2
回答已采纳
Kafka DirectStream的火花流问题：
1)调整spark.streaming.kafka.consumer.poll.ms增长2)将执行器内存从1G调整为2G。-部分工作，更别说失败了-当流持续时间小于8s
浏览 4提问于2017-02-16得票数 6
我正在使用Kafka流媒体从Kafka主题消费。(kafka直播流)该topic中的数据每隔5分钟就会从其他来源到达。现在我需要处理每隔5分钟到达的数据，并将其转换为Spark DataFrame。现在，流是连续的数据流。我的问题是，我如何确定我已经读完了加载到Kafka主题中的第一组数据？(这样我就可以将其转换为DataFrame并开始我<em
浏览 0提问于2017-06-09得票数 0
post将最大轮询记录设置为1，以保证一个流文件中的事件来自同一分区。https://community.hortonworks.com/articles/223849/simple-backup-and-restore-of-kafka-messages-via-ni.html这是否意味着如果使用消息分隔符，同一FlowFile中的事件可以来自不同的分区？https
浏览 11提问于2019-01-19得票数 0
回答已采纳
我对spark结构化数据流和kafka的偏移管理是新的。使用spark-streaming-kafka-0-10-2.11。在消费者中，我如何从主题的特定分区中读取？comapany_df = sparkSession .format("kafka").option("<e
浏览 22提问于2019-05-29得票数 0
2回答
对时间序列数据流执行查询
、、、
我正在尝试为我的流媒体应用程序设计一个架构，并为我的工作选择合适的工具。这是它目前的工作方式：
来自“应用程序生产者”部分的消息具有(address_of_sensor, timestamp, content)元组的形式。在Kafka之前我已经实现了所有的功能，现在我在设计中遇到了主要的缺陷。在“火花流”部分，将合并的消息流转换为事件流。问题是大部分
浏览 0提问于2015-11-04得票数 2
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
