我有一个对象事件的kafka流： KStream<String, VehicleEventTO> stream = builder.stream("mytopic", Consumed.with(Serdes.String我需要建立一个流，返回不同的idTypes到时间间隔(例如: 10分钟)。使用KafkaStream数字用户线是可能的吗？我找不到解决办法。
浏览 17提问于2019-06-29得票数 0
回答已采纳
在我的应用程序中，我将对我的Kafka集群执行某种健康检查。目前，我创建了一个TopicMetadataRequest来检测死掉的代理： // this will fail if Kafka is unavailable }
不幸的是，由于集群拓扑/设置<em
浏览 1提问于2015-07-17得票数 10
回答已采纳
电流配置
应用程序运行在应用程序中的Flink 1.14.4上，数据流(由长链操作符产生)间隔连接到一个Kafka源。事件时戳每个分区严格上升。卡夫卡上的每分区水印策略将水印设置到它迄今看到的最大时间戳(有界无序=1秒)。默认情况下， Kafka源(间隔连接的右侧)比其他数据流(kafka > source )每1分钟生成一次数据，因此它<
浏览 12提问于2022-06-17得票数 0
回答已采纳
我正在用Java/Python构建一个使用kafka和Kafka流的产品。
该产品将处理来自几个不同外部数据源的数据，这些数据源包括实时事件和每小时批处理事件。可能会有很长的差距，几个小时在一起，没有任何事件将来自外部来源。其中一些来自不同来源的事件需要进行关联和聚合，为此我计划使用Kafka流。鉴于Kafka/Kaf
浏览 1提问于2019-05-16得票数 1
2回答
我无法理解kafka流中的groupBy/groupById和窗口的概念。我的目标是在一段时间内(例如5秒)对流数据进行聚合。}下面是我正在尝试的代码，但我似乎无法让它工作。
builder.TimeVal tv = gson.fromJson(val, TimeVal.
浏览 0提问于2018-08-10得票数 7
我有一个结构化的流媒体应用程序读取来自Kafka的消息。每天的信息总数约为180亿条，每分钟的峰值信息计数为12,500,000条。最大消息大小为2KB。如何确保我的结构化流应用程序能够处理如此大的数据量和速度？基本上，我只想知道如何设置最佳触发时间、maxOffsetsPerTrigger或任何其他配置，从而使作业顺利进行，并能够处理故障和重新启动。
浏览 2提问于2019-06-17得票数 5
在我的WindowsMVC4项目中，我需要创建不同的作业进程来在不同的时间间隔执行。我计划使用ASP.NET Task Scheduler来实现这一点。
因为我是新手，所以我需要一些帮助。我的场景是我需要以特定的时间间隔向客户列表发送电子邮件。为了获取客户详细信息列表，我提供了web服务(它将提供选定客户的所有电子邮件i).These web服务
浏览 0提问于2014-02-10得票数 1
我正在用Apache阅读“流处理”一书，其中说：“从0.10.0版开始，Kafka支持消息时间戳。当从Kafka版本0.10或更高版本读取时，如果应用程序以事件时间模式运行，使用者将自动提取消息时间戳作为事件时间戳*“那么在processElement函数中，调用context.timestamp()将默认返回kafka消息时间戳吗？请提供一个简单的示例，说明如何实现基于使用的卡夫
浏览 4提问于2020-08-30得票数 2
回答已采纳
如果我是正确的，默认情况下，星火流1.6.1使用单个线程从每个Kafka分区读取数据，假设我的Kafka主题分区为50，这意味着每个50个分区中的消息将按顺序读取或可能以循环方式读取。创建多个KafkaUtils.createDirectStream是唯一的解决方案吗？案例2：
-If --我的卡夫卡分区每秒收到5条消息--那么，在这种情况下，"--conf spark.streaming.kafka.m
浏览 4提问于2016-12-12得票数 2
为什么你用卡夫卡来谈论数据流。也许有人能告诉我术语数据流在Kafka中的真正含义。顺便说一句，我知道在Kafka中有生产者和消费者。生产者将数据发送到代理，然后代理将其发送给消费者。
浏览 0提问于2021-05-02得票数 1
1回答
如何避免火花流中批量排队
、、、
我有直接流的火花流，我使用下面的配置
spark.streaming.kafka.maxRatePerPartition 42
我看到，最初的</
浏览 2提问于2021-03-16得票数 1
回答已采纳
是针对每个批处理间隔查询偏移量还是以不同的频率查询偏移量？
浏览 1提问于2015-11-17得票数 1
2回答
我们正在构建一个使用火花流和Kafka的容错系统，并且正在测试检查点火花流，如果火花作业因任何原因崩溃，我们可以选择重新启动它。下面是我们的火花过程：
卡夫卡每秒收到大约80条信息。我们想要实现的是一个设置，在这里我们可以降低火花流作业(以模拟故障)，然后重新启动它，并仍然确保我们处理来自Kafka的</e
浏览 1提问于2016-10-06得票数 2
2回答
我正在考虑Kafka从后端到前端应用程序的更新。数据流是特定于用户请求的，因此每个请求都会在后端生成一个流。-每个用户将有多个并发请求。一对多的关系，即用户和流在网上阅读时，我看到了一些帖子，其中提到了在userid上分区的一个主题。如果对userid进行分区是可行
浏览 1提问于2018-02-19得票数 1
回答已采纳
1回答
强迫汇流s3水槽冲洗
、、、
我设置了kafka连接s3接收器，持续时间设置为1小时，还设置了一个相当大的冲洗计数，比如10,000次。现在，如果卡夫卡通道中没有多少消息，s3接收器将尝试在内存中缓冲它们，然后等待它们累积到刷新计数，然后将它们一起上传，并将偏移量提交给自己的使用者组。
但想想这种情况。然后，经过很长一段时间，这5,000条信息最终将被驱逐出卡夫卡，因为保留时间。但是这些消息仍然存在于s3接收器的内存中，而不是在s3中。这是非常危险的，例如，如果我
浏览 3提问于2018-06-08得票数 4
回答已采纳
1回答
我在我的网站上注册网页浏览。我们已经对切分键进行了散列，以便在多个碎片中平均分配数据。然后，我们的聚合查询在间隔时间范围内运行，以聚合这些数据并提供站点上的趋势。我们遇到了Kafka，因为它是为重载和这种流编写发行的。
我对这两个系统进行了比较，并在主题中提供了分区上的分布，并采用了领导跟随方法。Kafka在具有分区复制的不同代理上使用多个分区，而Mongo则使用具有副本集<em
浏览 0提问于2018-03-21得票数 2
回答已采纳
我想从各种S3存储桶(超过10,000个存储桶，每个文件大约20-50MB)加载数据到Apache Kafka。存储桶列表是动态的-在运行时添加和删除存储桶。理想情况下，每个存储桶配置都应该有自己的轮询间隔(扫描新文件的频率-至少60秒，但可能更长)和优先级(正在处理的并发文件的数量)。请注意，由于每个存储桶所有者的组织中存在各种IT策略，因此无法设置从每个S3存储桶到SQS/ in &#
浏览 49提问于2021-07-28得票数 2
1回答
我正在尝试将数据写入一个卡夫卡主题后，阅读了一个蜂巢表如下。("spark.streaming.backpressure.enabled",”true”)
启用或禁用火花流的内部背压机制(自1.5起)。这使得火花流能够根据当前批处理调度延迟和处理时间来控制接收速率，以便系统接收到的速度仅限于系统所能处理的速度。在内部，这动态地设置接收器的最大接收速率。此速率是由spark.stre
浏览 0提问于2021-09-13得票数 4
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
云直播
对象存储
腾讯会议活动推荐
腾讯云开发者
