Kafka流DSL谓词流分离
是指在使用Apache Kafka流处理平台时，使用流式领域专用语言（Domain Specific Language，DSL）来处理数据流，并将数据流分为谓词流和非谓词流。
谓词流指的是根据某些条件进行筛选和过滤的数据流，可以根据特定的谓词（即逻辑条件）将数据流中的数据进行分类、过滤或聚合。谓词流通常用于对数据进行预处理、过滤无用数据或将数据按照特定条件分流。
非谓词流指的是不需要进行筛选和过滤的数据流，这些数据流中的数据是全部接收并处理的。非谓词流常用于对数据进行实时计算、聚合、转换或持久化等操作。
Kafka流DSL谓词流分离的优势在于：
- 灵活性：使用DSL可以快速、简便地定义和应用各种谓词和非谓词操作，提供了更灵活的数据处理方式。
- 效率：通过将数据流分为谓词流和非谓词流，可以减少不必要的数据传输和处理，提高处理效率。
- 可维护性：DSL可以使数据流处理逻辑更加清晰和易于维护，减少了代码的复杂性。
应用场景：
- 实时数据处理：Kafka流DSL可以用于处理实时生成的数据流，如日志数据、传感器数据等。
- 流式ETL：DSL可以用于将数据流进行转换、清洗和聚合，从而实现实时的数据提取、转换和加载。
- 实时计算和监控：通过DSL对数据流进行实时计算和监控，可以快速发现并处理异常情况或实时业务需求。
推荐的腾讯云相关产品和产品介绍链接地址： 腾讯云提供了一系列与流处理相关的产品和服务，以下是其中几个推荐的产品：
- Apache Kafka：腾讯云提供的高性能分布式流处理平台，可实现高可靠性、低延迟的数据流处理。 产品介绍链接：https://cloud.tencent.com/product/ckafka
- Tencent Cloud StreamSQL：基于Kafka的流式计算引擎，提供了SQL和DSL两种编程模型，支持实时数据处理和计算。 产品介绍链接：https://cloud.tencent.com/product/tcs
- Tencent Cloud TKE：腾讯云容器服务，可用于部署和管理Kafka流处理应用程序的容器集群。 产品介绍链接：https://cloud.tencent.com/product/tke
请注意，以上推荐的产品仅为示例，不代表其他厂商的产品。
相关·内容
1回答
使用Kafka流DSL 创建Kafka流时
我们遇到了一个场景，我们需要用新的拓扑定义更新Kafka流。
例如:当我们开始的时候，我们定义了一个拓扑来读取一个主题(来源)和一个目标主题(Sink)。但是，在进行配置更改时，我们现在需要从两个不同的主题(如果需要的话是两个源)进行读取，并将其写入单个目标主题。
从我们现在构建的内容来看，拓扑定义是硬编码的，类似于在中定义的。
问题：
是否可以以声明式的方式(例如Json或其他什么)定义拓扑，而不需要对拓扑进行编码？
是否有可能重新加载一个现有的Kafka流使用一个新的定义卡夫卡流拓扑？
对于上面提到的#2，Kafka Streams
浏览 0提问于2019-06-18得票数 1
回答已采纳
1回答
我能否在Kafka流中使用处理器api，并从一个kafka集群中的某个主题读取数据，并在另一个kafka集群中将数据输送到kafka主题中？
使用处理器api是可行的吗？我知道，这是不可能的流DSL！
浏览 0提问于2018-09-16得票数 0
回答已采纳
我有一个Kafka流DSL应用程序，我们要求只处理一次，同样，我也添加了配置。
streamConfig.put(processing.gurantee, "exactly_once");
我正在使用kafka 2.7，我有两个查询
exactly_once和exactly_once_beta有什么区别？
如何测试此功能，以确保只处理一次消息
谢谢!
浏览 13提问于2021-05-30得票数 0
回答已采纳
我有一个关于Json数据的Kafka主题，如
{"storeid": 17, "data": "blablabla"}
{"storeid": 17, "data": "blublublu"}
{"storeid": 18, "data": "bliblibli"}
我需要的是将新事件发布到另一个主题/流中，当1000个事件或5分钟(这是第一次)传递时，包含相同存储区的合并数据。预期结果将是：
{"storeid": 17, "
浏览 4提问于2022-06-26得票数 0
为了比较使用RocksDB或内存中的Spring Cloud Stream应用程序(使用Kafka)的性能，我想将RocksDB更改为内存中。不幸的是，我找不到任何解决方案。我已经为DSL API和处理器API ()找到了一个解决方案，但我不确定它是否适用于kafka-streams绑定器。有人能在这个问题上帮我吗？
向马丁致以最美好的问候
浏览 20提问于2019-07-10得票数 0
对于Kafka流，如果我们使用较低级别的处理器API，我们可以控制提交与否。因此，如果我们的代码出现了问题，并且我们不想提交这个消息。在这种情况下，卡夫卡将多次传递这条信息，直到问题得到解决。
但是，如何控制在使用其高级流DSL时是否提交消息？
资源：
浏览 2提问于2017-02-05得票数 10
回答已采纳
最近，在一次采访中，我被问到一个关于Kafka Streams的问题，更具体地说，面试官想知道为什么/什么时候你会使用Kafka Streams DSL而不是普通的Kafka Consumer API来读取和处理消息流？我无法提供一个令人信服的答案，我想知道其他使用这两种流处理风格的人是否可以分享他们的想法/意见。谢谢。
浏览 2提问于2020-09-13得票数 3
我目前正在使用Kafka流式DSL抑制功能的日期窗口。我们可能会遇到这样一种情况，一些事件可能会很晚才发生，超过了宽限期。
根据kafka流媒体文档，这样的事件将被丢弃，这不适合window。
请帮帮我。
1)是否有可能在同一流程中获得此类丢弃的事件？
Apache flink确实提供了对这种非常晚的事件的控制，并想知道这种功能是否在流中可用。
2)考虑到数百万事件流经系统，如何使用DSL- suppress将间歇性聚合数据保存在内存中？
任何timeline kafka流媒体社区都将很快提供rockDB支持，以避免应用程序因内存不足而崩溃。
浏览 2提问于2019-05-15得票数 0
1回答
我有一个Kafka流，它从一个主题中获取数据，并需要将该信息过滤到两个不同的主题。
KStream<String, Model> stream = builder.stream(Serdes.String(), specificAvroSerde, "not-filtered-topic");
stream.filter((key, value) -> new Processor().test(key, value)).to(Serdes.String(), specificAvroSerde, "good-topic");
stream.fi
浏览 0提问于2016-12-01得票数 6
回答已采纳
2回答
来自火花流的背景-掌握卡夫卡流。
我有一个简单的星火流应用程序从Kafka读取，
并在该分钟内返回每个用户的最新事件。
示例事件将类似于{"user": 1, "timestamp": "2018-05-18T16:56:30.754Z", "count": 3}, {"user": 1, "timestamp": "2018-05-22T16:56:39.754Z", "count": 4}
我感兴趣的是这将如何在Kafka流中工作，因为似乎每个事件都有一个输出--
浏览 0提问于2018-05-22得票数 1
回答已采纳
我的要求是使用kafka流DSL跳过或避免从输入主题接收的重复消息(具有相同的密钥)。
在发生故障时，源系统可能会向输入主题发送重复的消息。
流-
源系统->输入主题-> Kafka流->输出主题
目前，我正在使用flatMap来生成多个有效负载密钥，但是flatMap是无状态的，因此在接收到输入主题时无法避免重复的消息处理。
我正在寻找DSL，它可以跳过从输入主题接收到的重复记录，并在发送到输出主题之前生成多个键/值。
在这里，我认为正确的一次配置将有助于根据键从输入主题中删除接收到的消息，但是看起来它不起作用，也许我不理解使用一次。
你能给它加点灯吗？
浏览 1提问于2019-04-23得票数 1
回答已采纳
1回答
我试图在Flink中实现一个事件时时态连接。下面是第一个联接表：
tEnv.executeSql("CREATE TABLE AggregatedTrafficData_Kafka (" +
"`timestamp` TIMESTAMP_LTZ(3)," +
"`area` STRING," +
"`networkEdge` STRING," +
"`vehiclesNumber` BIGINT," +
浏览 4提问于2021-09-30得票数 0
回答已采纳
我们是否应该自己在Processor实现中调用processorContext.commit()？我的意思是在预定的Punctuator实现中调用commit方法，或者在process方法中调用。 在哪些用例中，我们应该这样做，我们需要这样做吗？这个问题涉及到带有transform()的Kafka DSL和处理器API。 似乎Kafka Streams自己处理它，而且调用processorContext.commit()也不能保证它会立即完成。
浏览 16提问于2019-01-07得票数 8
回答已采纳
我们有一个带有多个avroregistry的KafkaAvroSerde。在某个时候，serde尝试在1 url上注册模式时超时，但是由于它向流应用程序抛出IO异常，流线程关闭了。从kafka流应用程序的角度来看，这种类型违背了在创建avro serdes时支持多个urls的功能，因为DSL堆栈中出现的运行时异常将关闭stream Thread。几个问题：
有什么好办法处理这件事吗？
我们是否需要在应用程序逻辑中执行重试(当您简单地将主题具体化到商店中时，这可能是很棘手的)？
否则是否有一个avroserde包装器
可以用实际配置的avroRegistry urls重试吗？
当
浏览 0提问于2018-11-02得票数 2
回答已采纳
1回答
使用spring集成的Kafka dsl，我想知道为什么侦听器不接收消息？但是，如果我用带有KafkaListener注释的方法替换，相同的应用程序可以很好地使用消息。我缺了什么DSL？
不消耗的DSL代码：
@Configuration
@EnableKafka
class KafkaConfig {
//consumer factory provided by Spring boot
@Bean
IntegrationFlow inboundKafkaEventFlow(ConsumerFactory consumerFactory) {
Integ
浏览 0提问于2019-07-01得票数 1
回答已采纳
我使用了最新的，并在消费者配置上卡住了。
这是我的使用者配置：
@Configuration
public static class ConsumerConfiguration {
@Autowired
private KafkaConfig kafkaConfig;
private Log log = LogFactory.getLog(getClass());
@Bean
public ConsumerFactory<String, String> consumerFactory() {
Map<String,
浏览 2提问于2016-09-21得票数 3
回答已采纳
我正在构建一个非常简单的KafkaStreams演示应用程序，以测试一个用例。
我无法升级我正在使用的Kafka broker (目前版本为0.10.0)，而且有几条消息是由一个预-0.10.0生成程序编写的，所以我使用了一个自定义TimestampExtractor，在我的主类开始时将它作为默认配置添加到配置中：
config.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, GenericRecordTimestampExtractor.class);
当使用我的源主题时，这是非常好的。但是在使用聚合操作符时，我遇到了一
浏览 0提问于2018-01-24得票数 8
回答已采纳
我一直在使用Kafka Streams，并获得了一些基本的功能，但我在理解Kafka Streams DSL中的process方法时遇到了一些问题。具体地说： 我知道有两种使用Kafka Streams的方法，低级处理API和高级Streams DSL。对于较低的级别，您可以更明确地定义拓扑，命名每个节点等，而流DSL则将大部分抽象出来。 然而，高级流DS1具有称为process()的方法，该方法是终端操作(即，该方法返回void)。所以我的问题是，处理过的数据-Processor的void forward(key, value)方法发送的数据-去了哪里？ 在低级处理API中，您将自己命名为
浏览 9提问于2020-01-20得票数 0
回答已采纳
在Kafka流中重新划分一个大容量的主题可能非常昂贵。一种解决方案是通过生产者端的一个键对主题进行分区，并在Streams应用程序中摄取一个已经分区的主题。
有没有办法告诉Kafka流DSL，我的源主题已经由给定的键进行了分区，并且不需要重新分区？
让我澄清我的问题。假设我有这样一个简单的聚合(为了简洁而省略了细节)：
builder
.stream("messages")
.groupBy((key, msg) -> msg.field)
.count();
使用此代码，Kafka流将读取messages主题并立即将消息写回内部重新分区主题，这
浏览 2提问于2020-12-06得票数 1
回答已采纳
我需要在给定的时间范围内阅读Kafka主题之外的消息。我能想到的解决方案是首先找出时间范围开始的最大偏移量，然后继续消费消息，直到所有分区上的偏移量超过时间范围的结束。有没有更好的方法来解决这个问题？谢谢!
浏览 0提问于2014-07-16得票数 9
1回答
在阅读了很多关于数据流的内容后，我发现只有PubSub和bigquery是GCP数据流中支持流数据处理的数据源。 还支持哪些其他数据源？ 我是否可以使用数据流完成所有的流任务，这些任务可以使用其他ETL工具，如Spark或Kafka？
浏览 26提问于2020-07-01得票数 0
我有个问题。我无法找到通过过滤kafka文档的键来创建流的方法。
我希望过滤和操作kafka键的json，以检索以下示例的payload，该示例对应于我的couchbase id：
ksql>打印cb_bench_products-get_purge‘限值1；
Key format: JSON or HOPPING(KAFKA_STRING) or TUMBLING(KAFKA_STRING) or KAFKA_STRING
Value format: JSON or KAFKA_STRING
rowtime: 2022/03/04 10:49:43.643 Z, key: {&#
浏览 13提问于2022-03-04得票数 1
回答已采纳
在Samza和Kafka流中，数据流处理是在处理步骤( Samza中称为“作业”，Kafka Streams中称为"processor“)的序列/图(Samza中称为”数据流图“，Kafka Streams中称为”拓扑“)中执行的。在本问题的其余部分中，我将这两个术语称为工作流和工作者。
让我们假设我们有一个非常简单的工作流，包括一个工作人员A，它使用传感器测量值并过滤所有低于50的值，然后是一个工作人员B，它接收剩余的测量值，并过滤80以上的所有值。
输入(Kakfa主题X) ->(工作者A) ->(工作者B) -输出(Kafka主题Y)
如果我理解了
和
浏览 2提问于2016-12-09得票数 27
我正在用开发一个流应用程序。我的应用程序如下所示。
flatMap更改键并从单个消息生成多条消息。
使用步骤1中的键使用join和KTable。
transform用于有状态操作。
flatMap将密钥更改为原始密钥，即在步骤1之前。
groupBy具有键，如步骤4，实际上是步骤1。
reduce将记录“合并”到包含JSON数组的单个消息中。
净效果是将传入的消息(以key作为id1)拆分为多个消息(具有不同的键，例如k1、k2等)。使用join和transform增强每条消息。然后，将每条消息的密钥更改为id1。最后，将每个增强消息“合并”为一个消息，其中键为
浏览 10提问于2022-10-17得票数 0
1回答
我最近正在尝试使用kafka streaming进行一些敏感数据处理。我希望实现的目标是，当敏感数据被加密时，微服务架构的能力不会受到影响，即，紧密耦合的服务和流数据处理。 我的问题是，在kafka流中，有没有可能我用一个密钥解密传入的消息，然后用另一个密钥再次加密？我有一个计划，但由于我不熟悉kafka streaming，我不能证明kafka streaming有能力使用Streams DSL处理此功能。有没有人能帮我解决这个问题，最好告诉我Streams DSL中的哪个函数可以处理这个问题？ 更新:基本上，我问的是:我尝试对流管道中的单个消息使用公钥加密两次。一次在入站主题上，一次在出
浏览 34提问于2020-01-14得票数 0
回答已采纳
我在我的项目中定义了以下集成流程
///
public IntegrationFlow acarsEventFlow() {
return IntegrationFlows
//.from(Jms.messageDrivenChannelAdapter(this.acarsMqListener)) //Get Message from MQ
.from(org.springframework.integration.jms.dsl.Jms.messageDrivenChannelAdapter(
o
浏览 0提问于2019-03-14得票数 0
当使用Kafka流DSL时，是否有一种方法可以使用相同的主题作为两个不同处理例程的源？
StreamsBuilder streamsBuilder = new StreamsBuilder();
// use the topic as a stream
streamsBuilder.stream("topic")...
// use the same topic as a source for KTable
streamsBuilder.table("topic")...
return streamsBuilder.build();
上面的简单实现在运行时
浏览 0提问于2018-09-20得票数 5
回答已采纳
我想在一个Kafka流应用程序中同时使用处理器API和DSL。此外，如何在一个应用程序中构建和运行多个拓扑(比如1使用处理器API，其他使用DSL )。)
浏览 0提问于2019-04-20得票数 0
回答已采纳
使用spring集成Java dsl创建kafka使用者的java代码。
浏览 0提问于2018-08-01得票数 0
我正在使用apache camel集成我的kafka消息。此外，我还使用JAVA DSL来使用来自kafka端点的消息。
使用apache kafka API，它知道如何提交具有给定属性切换的消费者偏移量。
如果我在camel-kafka组件中取消了自动提交，那么我如何在apcahe camel中提交偏移量呢？
我使用下面的端点在Apache Camel中禁用自动提交
kafka.consumer.uri = kafka://{{kafka.host}}?zookeeperHost={{zookeeper.host}}&zookeeperPort={{zookeeper.port}}&
浏览 9提问于2015-09-21得票数 5
我在Kafka流中是新手，我的用例是将来自输入主题的每个单独传入记录的值与来自其单独先前记录的值进行比较，如果比较条件为真，则将包含比较结果的新记录与每个比较记录的索引发送到结果主题，否则不发送任何内容。(请注意，对于每个记录，所有传入记录都可能有一个唯一的键，也可能有一个空键)。
在Kafka消费者和生产者API中做这件事非常容易，但是((不使用外部DB来存储前面的记录)我只尝试使用(Kafka streams DSL API)，(其中包括KTable和KStream，以及它们的内部方法，如aggregate，reduce。等)，但可能是因为我是一个初学者，所以我没有在这些API中找到一种明
浏览 2提问于2018-12-23得票数 1
我正在使用spring集成dsl来接收来自Kafka的消息，然后在Oracle、Couchbase中解析、丰富和持久化，并将其发布到另一个Kafka主题以用于下行通道。
持久化和发布需要在一个事务中，以便所有数据源/数据存储同步。如果Couchbase、Oracle或Publish Kafka主题不可用，则回滚事务。
同时，我不希望在处理消息时出现延迟，因为这是对业务用户的实时更新。
return IntegrationFlows
// .from(Jms.messageDrivenChannelAdapter(this.acarsMqListe
浏览 25提问于2019-08-10得票数 0
1回答
Akka流中的协调流
、、、、
我们正在使用Kafka，并且希望使用来访问我们州商店中的数据。我们有一个现有的服务，它使用Akka HTTP来提供REST，我们希望将交互式查询集成到流中。
似乎是一个完美的适合这一点。但是，它通过公开一个使用低级别API的route属性来集成到Akka，该API映射到一个Flow[HttpRequest, HttpResponse, Any]。我们之前的所有代码都使用Akka HTTP的路由DSL连接代码。
我希望像下面这样的代码能够工作，但它不能：
implicit val system:ActorSystem = ActorSystem("web")
implicit v
浏览 1提问于2018-10-17得票数 0
回答已采纳
我有一个带有KTable的非常简单的spring引导项目，我想在application.yml中定制我的配置，但是配置似乎没有被应用。这是我的配置文件application.yml
spring:
kafka:
bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
streams:
application-id: ${APPLICATION_ID:train-builder-processor}
buffered-records-per-partition: 50
cons
浏览 10提问于2022-03-03得票数 1
回答已采纳
需要一些指导卡夫卡流分裂。
我有一个像这样的消息值字段
{"name": "val1", "role": "val2"}
消息的键是一个字符串字段，我们在这里不担心它。
在name字段中，我得到多个值，与/类似于这个{"name": "tom/dick/harry", "role": "manager"}，我希望能够用名称字段中的多个/分隔值检查流中的那些记录，然后根据这些值进行拆分或分支，并将每条消息发送到输出主题。因此，在本例中，基本上1条消息转到3条不同的消息：
浏览 2提问于2022-07-04得票数 -1
2回答
暂停流消耗
、、
我正在处理一个处理很少记录的应用程序。请求率约为每分钟2个电话。这些请求是为一组数据创建和更新的。这些要求包括送货保证、可靠交付、订购保证和防止任何消息丢失。
我们的团队已经决定使用Kafka，我认为它不适合用例，因为Kafka最适合流数据。相反，我们本来也可以更好地使用传统的消息模型。虽然Kafka确实为每个分区提供了排序，但在传统消息系统上，如果消息数量较少，数据源也较低，则可以实现同样的功能。这是一个公平的声明吗？
我们使用Kafka流来处理数据，而处理要求我们对外部系统进行查找。如果外部系统不可用，则当外部查找系统可用时，我们停止处理并自动向目标系统传递消息。目前，我们停止处理
浏览 10提问于2016-08-11得票数 1
回答已采纳
我有一个包含客户记录的卡夫卡主题，叫做“客户创造的”。每个客户都是该主题中的新记录。有4个分区。
我有两个基于docker映像confluentinc/cp-ksql-server:5.3.0的ksql服务器实例正在运行。两者都使用相同的。
我创造了一张桌子：
CREATE TABLE t_customer (id VARCHAR,
firstname VARCHAR,
lastname VARCHAR)
WITH (KAFKA_TOPIC = 'customer-created&#
浏览 1提问于2019-08-02得票数 1
回答已采纳
这个问题是的后续问题。我需要将小字典数据加入到主要的Kafka流中(就像“地图端”连接一样)。
AFAIK，Kafka实例总是在主题的给定分区上工作。如果我想进行查找，我需要为连接键重新划分这两个流，以便将相关记录放在一起。
如果需要检查多个查找数据，多次重新分区的成本是多少？不可能将整个查找数据集发送到每个分区，因此当我从查找主题构建KTable时，我将在所有Kafka应用程序实例中看到整个数据集。因此，我可以在KStream#transform()方法中进行查找，这将使用本地RocksDB存储区和我拥有的所有查找数据。
我想知道哪一种选择更合适：
向主题的每个分区插入相同的数据(整个数
浏览 1提问于2016-09-21得票数 4
回答已采纳
1回答
我们正在尝试用下面的代码实现Spring DSL Kafka的手动提交。我们找不到任何与之相关的资料。
我们可以添加"Auto.commit“作为"False”的消费者属性，但我们希望在成功处理消息后提交消息。有人能帮上忙吗？
@Bean
IntegrationFlow consumer() {
KafkaHighLevelConsumerMessageSourceSpec messageSourceSpec = Kafka
.inboundChannelAdapter(new ZookeeperConnect(
浏览 0提问于2016-05-09得票数 1
我有下一个主题的值
// photos
1, {"id": 1, user_id: 1, "url": "http://example.com"}
2, {"id": 2, user_id: 2, "url": "http://example1.com"}
3, {"id": 3, user_id: 1, "url": "http://example2.com"}
// users
1, {"id": 1, "name"
浏览 0提问于2018-07-08得票数 1
回答已采纳
我有以下处理器bean方法签名：
@Bean
public BiFunction<KStream<String, MyClass>, KStream<String, String>, KStream<String, MyClass>[]> myStream() {
return (inputStream1, intputStream2) -> {
intputStream2
.peek((k, v) -> {
log.debug(...);
浏览 1提问于2020-07-09得票数 0
回答已采纳
我正在尝试将一组广泛的路由配置转换到Spring DSL。
我遇到的一个问题是(非常常见的情况)，多条路径应该映射到同一条路由。也就是说，任何匹配/abc/**、/def/**或/ghi/**的传入URL都应该路由到某些uri http://example.org (并应用了相同的筛选器等)。
在基于Fluent Java的路由方法中，这可以通过.or().方法(例如：r.path("/abc/**").or().path("/def/**").or().path("/fhi/**")... )来实现。
如何通过基于YAML的路由配置实现相同的功
浏览 1提问于2018-08-24得票数 2
在kstreams High level DSL中，我们有to方法，有没有一种方法可以根据输出消息中的键，以分区的方式将输出写入kafka。
浏览 1提问于2017-07-01得票数 0
回答已采纳
我正在编写一个通用适配器，从EMS获取消息，并推送到Kafka主题。如果配置文件中的某个特定标志设置为true，则我必须丰富标头。标头的表达式也将存在于配置文件中 return IntegrationFlows.from(org.springframework.integration.jms.dsl.Jms
.messageDrivenChannelAdapter(org.springframework.integration.jms.dsl.Jms
.container(this.queueConnFactory, this.queue)
浏览 7提问于2020-04-07得票数 0
我想连锁一堆过滤器，但不想要与创建多个列表相关的开销。
type StringFilter = (String) => Boolean
def nameFilter(value: String): StringFilter =
(s: String) => s == value
def lengthFilter(length: Int): StringFilter =
(s: String) => s.length == length
val list = List("Apple", "Orange")
问题是，这会在
浏览 1提问于2015-02-20得票数 4
回答已采纳
我正在尝试在这里构建Materialized.as DSL代码：https://kafka.apache.org/11/javadoc/org/apache/kafka/streams/state/Stores.html 但是我得到了一个错误 incompatible types: org.apache.kafka.common.serialization.Serde<java.lang.Long> cannot be converted to org.apache.kafka.common.serialization.Serde<java.lang.Object>
浏览 33提问于2019-01-06得票数 1
回答已采纳
Spring boot Apache Camel-Java DSL应用程序从Kafka主题读取消息。 @Component
public class KafkaTopicService extends RouteBilder {
public void configure(){
from("kafka:myTopic?brokers=localhost:9092")
.log("Message received from Kafka: ${body}")}
} 如果我阻止卡夫卡我会得到org.apache.kafk
浏览 66提问于2020-11-29得票数 0
当源主题分区计数= 1时工作正常。如果我将分区增加到任何大于1的值，我会看到下面的错误。既适用于低级，也适用于DSL API。有什么建议吗？可能会遗漏什么？
org.apache.kafka.streams.errors.StreamsException: stream-thread [StreamThread-1] Failed to rebalance
at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:410)
at org.apach
浏览 3提问于2017-02-19得票数 6
回答已采纳
问题是-我有一个程序可以构建和运行良好。它是用Scala编写的，它使用Kafka流DSL。我本来要使用新的kafka-streams-scala包，现在我正在使用Maven。我现在不能用SBT。当我添加依赖项时
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-streams-scala_2.11</artifactId>
<version>2.0.0</version>
浏览 0提问于2018-09-12得票数 0
回答已采纳
在网上读了很多之后，我终于找到了这个论坛。我面临的挑战是，在将事务数据写入NoSQL数据库(在本例中为Cassandra )之前，将来自CDC的数据库中的事务数据反规范化到Kafka中。将事务数据与主表中的查找连接起来的最佳方式是什么？我的问题是，每个事务表可能有5到10个查找表。
在使用KSQL进行概念验证时，我学到了以下几点: A)将查找表作为KTables加载，B)对事务流进行重新分区，最后C)执行连接并写入新主题。按照这种方法，如果我有5个或10个查找表，它们将生成在集群中发送的大量数据。我知道Streams DSL可以使用GlobalKTable的概念，但只有在查找表相对较小的情况下
浏览 1提问于2019-10-10得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
