0回答
我一直在使用Apache Kafka实现ETL数据管道。我已经使用Kafka Connect进行提取和加载。
Connect将读取源数据，并以JSON的形式写入Kafka topic的实际数据。在转换阶段，我想从Kafka主题读取JSON数据，然后需要转换为基于一些自定义业务逻辑的SQL查询，然后需要写入输出Kafka主题。到目前为止，我已经编写了一个生产者-消费者应用程序，它从主题读取，进
浏览 3提问于2017-12-15得票数 2
回答已采纳
我正在写一个使用Kafka streams的应用程序。它从主题A读取，进行一些转换，并写入主题B。在转换期间，值按键分组，因此输出键、值类型与输入值类型不同。Kafka streams使用特定类型的Serdes (例如String serdes序列化和反序列化字符串)进行序列化和反序列化，因此在数据转换后它将不起作用。如何在Streams API中定义不同的序列化程序和反序列化程序
浏览 15提问于2018-07-27得票数 0
回答已采纳
1回答
-1e64cf93-de19-4185-bee4-8fc882275010-StreamThread-1" org.apache.kafka.streams.errors.StreamsException:117)
at org.apache.kafka.streams.processor.internals.StreamTask.addRecords(StreamTask.java:677)at org.apache.kafka.streams.processor.i
浏览 4提问于2019-11-13得票数 0
示例：此消息结构在kafka主题中，但我希望在elasticsearch中对其进行索引，如下所示
{"id":"1","name"
浏览 0提问于2018-09-20得票数 0
我需要写一个java应用程序，从Kinesis流读取和存储数据到Kafka主题。我不能使用连接器，我必须使用Kafka Streams，因为其中涉及到很多逻辑。Kafka Stream是否可以直接从Kinesis流读取，或者我必须先将数据存储到Kafka topic才能被Kafka Stream读取？我读了很多关于Kafka Streams的文章，除了Kafka主题之外，我从来没有见过Kaf
浏览 18提问于2019-07-29得票数 1
因此，我已经实现了一个自定义SerDe，它扩展自定义的SpecificAvroSerde，在与Schema进行超时通信时尝试重试。(StreamThread.java:951) ~[kafka-streams-2.5.0.jar:na]
at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:697) ~[kafka-<
浏览 3提问于2020-12-14得票数 0
回答已采纳
我们有发布到Kafka主题的学生数据流，我们的模块必须以流的形式读取学生ids，并从多个来源获取每个学生的关联数据，然后为每个学生执行一些计算，并将每个学生的关联计算发布到kafka主题中。因此，这里的问题是更好地编写一个大的Spark作业或使用Akka为每个源提供单独的服务，以便参与者可以并行工作，获取一堆学生it，并从各自的源获取数据，并执行一些簇转换和操作，最后执行与每个学生相关的计算或者我真的需要在这里使用Akka？Spark会在内部有效地处理这个问题吗？
感谢你在这里提出的任何想法。
浏览 2提问于2017-07-14得票数 1
1回答
我在一个微服务架构中使用flink streams将数据从服务ETL到报告和搜索数据库。为了感知数据变化，我目前正在为RMQ生成自定义事件，然后使用streams根据报告和搜索需求对数据进行分区、聚合和转换。 我迁移到kafka作为代理和debezium，从所有微服务的数据库中获取数据。我想知道我是否可以使用KStreams应用程序接口或flink，或者是否有任何可能性将两者结合起来。
浏览 14提问于2019-01-24得票数 1
问题描述：我的问题是：任何想法都很感激。应用程序Config：
浏览 8提问于2019-10-21得票数 0
回答已采纳
我是kafka社区的新手，我面临着一个具有挑战性的问题。我有两个应用程序，它们通过Rest webservice相互通信，主体是一个json消息。我如何使用kafka作为这两个应用程序之间的中间件，而对应用程序的影响很小或为零？ 这是我的原样场景： ? 我的未来场景： ? 我考虑过使用STM或一些拦截器来转换头部和正文。{ "last name"
浏览 35提问于2019-01-25得票数 2
回答已采纳
我们有以下依赖关系： libraryDependencies += "org.apache.kafka" %% "kafka-streams-scala" % kafkaVersion1.4.0"
libraryDependencies += "com.sksamuel.avro4s" %% "avro4s-core" % "3.0.4" 我们使用代
浏览 22提问于2020-04-08得票数 2
回答已采纳
._import org.apache.kafka.clients.consumer.ConsumerConfigimport org.apache.kafka.common.serialization.word
浏览 1提问于2020-03-21得票数 1
回答已采纳
我使用Debezium MySql连接器将数据库更改事件从MySql数据库流式传输到Kafka。我需要将特定的转换应用于某些指定表(而不是其他表)中的记录。有没有一种只使用Kafka Connect单消息转换(没有Kafka Streams或KSQL)的方法？我甚至不介意写一个自定义的SMT。但我找不到如何为消息指定要应用SMT的表/主题。谢谢。在简短的调查之后，我决定使用KSQL。感谢大家的回答
浏览 10提问于2019-04-11得票数 1
我试图使用kafkastreams进行聚合，得到了如下所示的错误KGroupedStream<String, Long> countrywiseAmount = ......;在org.apache.kafka.streams.processor.internals.SinkNode.process(SinkNode.java:94) at org.apache.kafka.streams.processor.internals.ProcessorCo
浏览 0提问于2020-06-24得票数 0
回答已采纳
我有一个流应用程序与下面的驱动程序代码的实时消息转换. at org.apache.kafka.streams.processor.internals.AbstractTask.我注意到的一件事是，应用程序可以正常工作，直到它转换所有的待办事项，但在处理一些新消息后抛出异常！
有时，它也抛出以下未明的异常。UncaughtExceptionHandler stream-thread [StreamThread-17
浏览 17提问于2017-02-22得票数 7
回答已采纳
我使用1.1.0。readOnlyKey: String, value: String): Test = Test("aaa", 432)我想构建一个可查询的存储，以后可以用来查询它(检索、过滤/转换的值我使用TopologyTestDriver运行了一个简单的测试，并引发了以下异常：
原因: org.apache.kafka.common.serialization.StringSerializer.serialize:66) at org.apache.<e
浏览 2提问于2018-12-27得票数 1
回答已采纳
1回答
我正在寻找该窗口的结果，并将其再转换一次，以获得我想要推入水槽的最终结果。这些是我的DSL调用来完成我的聚合。(ProcessorStateManager.java:242)我不确定异常的确切原因，但是，我确实认为存在一个问题，即进入最后一个转换的流的关键是一个Windowedarguments (org.apache.kafka.streams.kstream.KeyValueMapper[org.apache.kafka.streams.kstrea
浏览 0提问于2018-04-02得票数 0
回答已采纳
KeyValue<>(key, outputMessage); } cloud: kafka: binder: - ${NX_KAFKA_SERVERS}
bindings我正在使用Jackson序列化程序。如何创建inputTopic和output
浏览 6提问于2020-10-09得票数 0
回答已采纳
这是关于升级生产中的现有代码库，这些代码库使用来自kafka-clients，kafka-streams，spring-kafka 2.4.0到2.6.x的窗口，并将spring-boot-starter-parent2.2.2.RELEASE升级到2.3.x，因为2.2与kafka-streams 2.6不兼容。streams code}
现在，将kafka streams</em
浏览 20提问于2021-05-15得票数 2
在自定义Transformer中，我们尝试在init(..)方法中创建和转发一些记录。根据Kafka Streams JavaDoc的说法，这应该是可能的。:517)at org.apache.kafka.streams.processor.internals.Assign
浏览 7提问于2019-03-16得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
腾讯会议
云直播
对象存储活动推荐
腾讯云开发者
