我是kafka流处理器的新手，偶然发现了“拓扑”的关键概念。我创建了一个源代码处理器，它从一个" source -topic“中读取数据，如下所示： Topology topology = new Topology();
topology.addSource("SOURCE", "source-topic"); 上面的代码片段将创建(如果我的理解正确的话)一个名为" source“的源流处理器，并将侦听kafk
浏览 32提问于2019-03-19得票数 1
回答已采纳
它只为kafka流DSL提供数据/主题管理。有人能为kafka流的处理器API共享任何相同类型的数据管理链接吗？我对处理器API的用户和内部主题管理特别感兴趣。TopologyBuilder builder = new TopologyBuilder();
// add the source processor node that takes Kafka topic"source-to
浏览 5提问于2017-10-26得票数 0
回答已采纳
i使用的是卡夫卡处理器api (而不是DSL)
{
String[] topicList = stateStore.get(key).split("|"); } 场景：我们从源主题读取数据，
浏览 1提问于2020-06-22得票数 0
回答已采纳
使用kafka流处理器api topology.addSink("sink2","output_topic2","StreamsProcessor"
浏览 2提问于2020-06-26得票数 0
2回答
当在kafka流中定义拓扑时，可以添加一个全局状态存储。它将需要一个源主题以及一个ProcessorSupplier。处理器接收记录，并在将它们添加到存储区之前从理论上对它们进行转换。但是在恢复的情况下，记录直接从源主题(changelog)插入全局状态存储，跳过最终在处理器中完成的转换。根据文件
注意:您不应该使用处理器将转换后的记录插入到全局状态存储中。此存储将源主题用作changelog，并且在还原期间将直接从</em
浏览 2提问于2019-11-25得票数 18
回答已采纳
1回答
我是新开发卡夫卡流应用程序。我的流处理器是根据输入json消息中的用户键的值对json消息进行排序。在kafka中，所有不同的流处理器应用程序都是从同一个json输入主题中读取的，但每个应用程序只在满足预设用户条件的情况下将消息写入特定用户的输出主题。KafkaStreams streams = new KafkaStreams(builder, props); }问题1:如果使用低级别<e
浏览 2提问于2017-05-07得票数 3
回答已采纳
2回答
我创建了一个类似于":test-topic > log“的流。我可以看到日志接收器使用主题"test- topic“中的数据。但是，如果我向SCDF.Then添加一个自定义处理器，我会创建一个如下所示的流：我认为"etl- data -transform“处理器会消耗测
浏览 1提问于2021-04-02得票数 0
1回答
我正在为数据源-处理器-接收器的数据流模型寻找azure替代方案。我希望这三个实体是独立的微服务。我想使用消息传递作为这三者之间的纽带。基本上，源应用程序从另一个服务获取数据，并将其发送到处理器，而处理器应用程序对其进行操作，并向宿发送相关的通知/警报。我该如何使用它们？
浏览 2提问于2018-01-05得票数 0
作为应用程序逻辑的一部分，我们使用Kafka流状态存储进行范围查找，使用builder.table()方法从builder.table()主题加载数据。但是，看起来streams实现重置了传递给原始serdes的任何内容，这些serdes用于从表主题加载。这就是我在流构建器内部所看到的：
public synchronized <K, V> KTable<K, V> table(final String topic,请不要建议使用处理器A
浏览 0提问于2018-05-11得票数 2
我希望使用NiFi rest将流文件传输到输入端口。我正在浏览NiFi rest文档，发现似乎有实现相同目标的支持。然而，诚实并不是很清楚，也不是很详细。我的工作流程如下：源处理器组中的：
目标处理器组中的：
我想知道的是如何将排队的流文件从源处理器组传输到目标处理器组的输入端口，以便使用PutFile存储。我想尝试到达这个端点：http://localhost:80
浏览 2提问于2017-04-05得票数 0
对于部署，将有多个源输入到这个管道中，我可以使用数据流标签来完成这一工作。一切都很好。每个源都是不同的rabbitmq实例，并且由于处理器需要知道消息来自何处(因为它必须调用源系统以获得进一步的信息)，我想出的策略是用源系统的头部细节充实每条消息，这些信息随后透明地传递给处理器。现在，我精通Spring、Spring和Spring，但我无法了解如何在数据流源组件中充实每条消息。<e
浏览 6提问于2016-11-23得票数 0
1回答
对于Kafka流，如果我们将commit.internal.ms配置设置为Long.MAX_VALUE，这将有效地避免卡夫卡流自动提交，但只有在调用低级处理器api context.commit()之后才能提交如果是，那么在出现问题或消息包含格式错误时，我们如何从主题中删除这些消息？
资源：
浏览 5提问于2020-08-13得票数 0
回答已采纳
我们正在使用低级处理器API开发一个Kafka Streams应用程序。 根据Kafka上的文档，所有的线程和并行性都是由Stream线程和流任务处理的。使用主题上的分区，并行性也是可扩展的。这意味着使用Kafka Streams API主要用于消费主题中的数据，而不是用于实际处理。实际处理将在Kafka流线程的初始数据消耗之后调用的新线程中进行。拓扑中的示例处理器： public class Processor implements Processo
浏览 15提问于2019-05-29得票数 2
出于各种原因，我们需要使用处理器api。
一个具有多个源主题的处理器。两个处理器，每个处理器都有自己的源主题。每个处理器只能读取对其他处理器状态存储的访问(如果可能的话)。也许可以解释一下dsl是如何实现的？
浏览 0提问于2018-11-26得票数 3
回答已采纳
我想知道是否有可能使用预设模块在spring cloud数据流中实现这样的流。处理器1、2、3都是预置模块(httpclient等)。处理器1和3将同时从源获得相同的消息，然后处理器1通过管道将输出传递给处理器2。当处理器2和3完成时，它们同时将输出通过管道传递给处理器4。我真的不关心处理器2的输出，处理器4只从3获得输出是很好的，但是处
浏览 19提问于2021-09-13得票数 1
回答已采纳
当使用Kafka流DSL时，是否有一种方法可以使用相同的主题作为两个不同处理例程的源？StreamsBuilder streamsBuilder = new StreamsBuilder();
streamsBuilder.stream("topic")...
streamsBuilder.tab
浏览 0提问于2018-09-20得票数 5
回答已采纳
对于Kafka流，如果我们使用较低级别的处理器API，我们可以控制提交与否。因此，如果我们的代码出现了问题，并且我们不想提交这个消息。在这种情况下，卡夫卡将多次传递这条信息，直到问题得到解决。但是，如何控制在使用其高级流DSL时是否提交消息？
资源：
浏览 2提问于2017-02-05得票数 10
回答已采纳
1回答
我正在尝试使用带有GetRethinDB处理器的apache nifi从我本地的rethinkdb中获取一些数据。它向我显示配置错误，即上游连接设置无效。有人能帮我理解它的含义和克服它的方法吗？有没有其他处理器可以和它一起使用？尼尔
浏览 0提问于2017-11-01得票数 0
使用kafka处理器API(而不是DSL)读取源主题并将其写入目标主题它适用于单个kafka集群设置(也就是说，如果源主题和目标主题驻留在同一集群上)，但是当源主题和目标主题驻留在不同的kafka集群上时topology.addProcessor("TargetProcessor",()->new TargetProcessor(), "Target");
topology.addSink("sink1","OUTPUT_
浏览 1提问于2020-06-18得票数 0
在spring运行时，可以让kafka源模块作为处理器模块工作吗？有代码样本吗？我正试图实现这样的目标: http (xd源)、xd源(xd处理器)、xd使用者(xd接收器)stream create kafkaSourceTest --definition "http --outputType=application/json | kafka
浏览 3提问于2015-05-12得票数 0
回答已采纳
云服务器
对象存储
ICP备案
云点播
实时音视频
腾讯云开发者
Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有
深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 
粤公网安备44030502008569号
腾讯云计算（北京）有限责任公司 京ICP证150476号 | 京ICP备11018762号