假设在此窗口中累积了数亿条记录，加入的消费者需要多长时间才能重启？我之所以这样问，是因为我无法找到有关窗口中有多少记录存储在使用者缓存中的信息。
浏览 7提问于2020-01-03得票数 0
回答已采纳
我们正在尝试实现下面描述的用例，我们有我们希望克服的实现问题，我们正在尝试通过匹配两个流的消息中存在的关键字(JSON)来在两个Kafka主题之间进行KStream连接。此外，我们应该维护消息序列，因为它是从源文件到达KStream的。问题1:如何在其他主题中停止或保持join，直到预期的键到达。例如，KTab
浏览 2提问于2019-05-08得票数 2
1回答
<VO，VR>join(org.apache.kafka.streams.kstream.KStream<java.lang.String，VO>，org.apache.kafka.streams.kstream.ValueJoiner扩展VR>join(org.apache.kafka.streams.kstream.KStream<java.lang.String，，org.a
浏览 4提问于2022-04-19得票数 1
回答已采纳
[_, _ <: VR])org.apache.kafka.streams.kstream.KStream[Nothing,VR] exist so that it can be applied to[AnyRef,Double]
required: org.apache.kafka.streams.kstream.ValueMapper[_, _ <: ?found : org.apache.kafka.streams.kstream.ValueM
浏览 4提问于2017-06-24得票数 4
回答已采纳
1回答
-repartition]) Source: KSTREAM-SOURCE-0000000028 (topics: [KSTREAM-PEEK-0000000025-repartition]) Processor: KSTREAM-JOIN-0000000009 (stores-repart
浏览 2提问于2019-11-22得票数 0
1回答
Kafka流我正在尝试流式传输，但我有一些问题，它不工作。首先，我有三个连接器，但我不能使用我自己的键。我需要钥匙才能加入他们对吧？如何使用2个或2个以上的密钥加入？value.converter=org.apache.kafka.connect.json.JsonConverter StreamsBuilder builder = new StreamsBuilder();
浏览 7提问于2020-01-15得票数 3
;import org.apache.kafka.streams.kstream.JoinWindows;import org.apache.kafka.streams.kstream.KStream;
importprops = new
浏览 0提问于2020-05-02得票数 2
到目前为止，我的代码如下： KStream<String, RecordTwo> recTwoStream = streamsBuilder.streamKStream<String, JoinedRecord> joinedRec
浏览 2提问于2022-10-23得票数 0
回答已采纳
> inputSchool1KStream();
}Exception in thread: class kafka.streams.join.School cannot be cast to class kafka.streams.join.Person (kafka.streams.join.Schooland
浏览 0提问于2019-06-11得票数 4
回答已采纳
我希望创建一个KTable (将来将是一个GlobalKTable)，以跟踪使用Streams通过我的Kafka代理程序的条目。我现在拥有的代码应该在侦听器透视图上正常工作，因为它接收通过Kafka传递的数据，并提取创建注册表所需的两个it。我在网上找到的所有示例都使用KStream/KTables作为方法的入口点/参数，但在某些方面没有使用其他类型条目的示例。
新的卡夫卡/春天云流。谢谢你的帮助！
浏览 0提问于2019-08-09得票数 2
回答已采纳
1回答
[org.apache.kafka.streams.state.KeyValueStore[_, _]])org.apache.kafka.streams.kstream.KTable[String,VR] <and>
[error] [VR](x$1: org.apache.kafka.streams.kstream.Initializer[VR], x$2: org.apache.kafka.streams.kstream.Aggregator[_
浏览 10提问于2018-01-08得票数 1
我试图用kafka流编写一个简单的过滤逻辑，但每次使用selectKey方法都会出错，无法进一步处理。value.split(",")(0).toLowerCase)overloaded method value selectKey with alternatives: (x$1: org.apache.kafka.streams.kstream.KeyValueMapper[_ >: String, _ >: String, _ <: String],x$2: org.apache.
浏览 9提问于2020-04-20得票数 0
回答已采纳
1回答
order: KStream<Int, Order>): KStream<Int, OrderShipped> {
.join(customerTable, { orderIt, customer -> stream:
浏览 1提问于2019-04-17得票数 0
回答已采纳
我有一个卡夫卡流应用程序，它接收来自主题-1作为KStream和主题-2作为KTable的数据。这两个主题都有4个分区。假设我有4个应用程序实例正在运行，那么每个实例将从主题-1的单个分区接收数据。如果这两个主题都是相同的，那么我猜Kafka流将确保为应用程序分配相同的分区。如果Top-2没有任何键，而是应用程序将从值本身推断出这一点，那么这意味着所有实例都需要从Top-2获取所有分区。
浏览 0提问于2018-04-27得票数 5
回答已采纳
2回答
我有3个不同的主题，在模式注册表中有3个Avro文件，我想将这些主题流到一起，并将它们写入到一个主题中。问题是，我想加入的键与我将数据写入每个主题的键不同。报警 "type" : "record", "namespace" : "com.kafkastream.schema.avro", "name" : "alarm_id"
浏览 0提问于2017-01-23得票数 2
回答已采纳
schema.registry.url: http://kafka-schema-registry.us-east-1.stage.kafka.away.black:8081percentiles = {0.5, 0.9, 0.99}, histogram = true) @SendTo("bridge")
public KStream<Windowed<String>, ResultValue> process(@Inp
浏览 0提问于2019-05-17得票数 1
1回答
var streamsBuilder = new StreamsBuilder(); inputStream.mapValues((ValueMapper更新:这是我可能的解决方案：public Consumer<KStream<
浏览 5提问于2021-07-20得票数 2
回答已采纳
0回答
我目前正在尝试使用KStream to KTable join来丰富Kafka主题。对于我的概念证明，我目前有一个包含大约600,000条记录的Kafka Stream，所有记录都具有相同的键，以及一个从具有1条记录的键、值对的主题创建的KTable，其中KTable主题中的键与创建KStream当我使用left join (通过下面的代码)时，ValueJoiner上的所有记录都返回NULL。StreamsConfig.NUM_STREAM_THREADS_CONFIG, 5);
浏览 2提问于2018-07-20得票数 0
回答已采纳
1回答
我在流上聚合值，如下所示：
private KTable<String, StringAggregator> aggregate(KStream<String, String> inputStream)
浏览 0提问于2018-06-06得票数 0
回答已采纳
.]);
KStream<String, RecordA> recordsAStream = builder.stream(= null) .leftJoin(mergedTable= null)
// <e
浏览 0提问于2019-05-21得票数 2
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
腾讯会议
云直播
对象存储活动推荐
腾讯云开发者
