我正在运行一个简单的Kafka streams应用程序，该应用程序使用Node JS将信息记录到Kafka主题。 E.g.我在scala中使用Kafka Stream来使用这些数据。TopicTest") println(k)} 但是，我不确定如何将时间戳(我从nodeJS发送的)提取到这个流中。Create Kafka</
浏览 41提问于2020-09-02得票数 0
= toTimestamp(value) key -> value}
}
其中，toTimestamp只是一个函数，它返回从记录值中获取的时间戳。6f3693b9-4e8d-4e65-9af6-928884320351-StreamThread-5" java.lang.NullPointerExc
浏览 1提问于2018-11-06得票数 2
回答已采纳
在下面的程序中，我试图更改值的情况，但它不起作用(val wordCountInputProcessed = wordCountInput.mapValues(value => value.toLowerCase卡夫卡流版本=> 2.3.0import java.util._import org.apache.kafka.cli
浏览 1提问于2020-03-21得票数 1
回答已采纳
我有3个数据源，它们提供了访问它们的api，每个数据源都提供了一些信息，我想处理所有这三个请求，并将它们保存在数据库中的单个请求中。所以我的问题是，如果所有这些源都有一个公共ID，并将其保存在数据库中并向下游系统发送单个请求，Kafka是否可以合并这些请求？
浏览 42提问于2019-05-06得票数 0
回答已采纳
我试图使用Scala中的ProcessorAPI运行一个非常基本的流。GenericRecord val currentState = Option(kvStore.get(key)) // NPE }似乎从错误日志中抛出了一些内部(MeteredKeyValueStore.java:134)
at writeside.kafka.AggregateKafkaProcessor.process(Kaf
浏览 1提问于2019-01-09得票数 0
回答已采纳
我有一个需求，我将WebSocket作为数据源，并希望将数据从WebSocket推送到Kafka topic中进行进一步处理。我不确定scala中是否有实现它的apis。
浏览 2提问于2019-04-04得票数 0
我刚接触Apache Kafka和它的与API相关的流服务，但想知道是否有任何正式的文档来获取摄取所需的初始原始数据？从本质上讲，我想尝试构建一个初级的密码交易机器人，但我的印象是http API可能比与Kafka Streams集成的API具有更多的延迟。例如，我知道RapidAPI有一个可以访问的http库，可以帮助提取数据，但我不确定如果我想通过Kafka Streams摄取数据，是否有类似的东西。我试着在Google上挖掘，但不是很清楚Kafka</em
浏览 0提问于2021-06-05得票数 0
我正在使用Spark2.2.0和Kafka0.10火花流库从满是kafka应用程序的主题中读取。Kafka Broker版本为0.11，Kafka-streams版本为0.11.0.2。当我在Kafka应用程序中设置EXACTLY_ONCE保证时：
p.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE当我运行kafka.tools.Ge
浏览 0提问于2018-01-19得票数 2
回答已采纳
1回答
我尝试了许多不同的选择变体，但应用程序运行，但没有显示消息，这些信息正在编写的每一秒。我有一个使用pprint()的星火流示例，它确认kafka实际上每秒钟都会收到消息。Kafka中的消息是JSON格式的，请参见字段/列标签的架构：from pyspark.sql.types import * .config("spa
浏览 6提问于2019-12-31得票数 4
回答已采纳
我们有以下依赖关系： libraryDependencies += "org.apache.kafka" %% "kafka-streams-scala" % kafkaVersion基本上，我现在得到了这个错误： could not find implicit value for parameter consumed: org.apache.kafka.streams.scala.kstream.Co
浏览 22提问于2020-04-08得票数 2
回答已采纳
假设我通过kafka streams从Confluent创建了主题，其中包含使用io.confluent.kafka.streams.serdes.avro.SpecificAvroSerializer在avro中序列化的消息CREATE EXTERNAL TABLE k_tableSTORED` > 1000 * t
浏览 2提问于2019-02-06得票数 0
我试图理解Kafka Streams API是如何与Schema Registry一起工作的。我知道在设置应用程序时必须指定Schema Registry URL，但我无法理解如果不指定主题名称或ID，应用程序如何从注册表检索正确的模式。
它使用主题名称检索模式？
浏览 2提问于2019-11-25得票数 2
我一直在下面的代码中获得‘歧义隐式值’消息。我尝试了几种方法(从我注释掉的几行代码中可以看出)。有什么办法解决这个问题吗？这是在Scala中。[T]
match expected type org.apache.kafka.common.serialization.Serde[org.apache.kafka.streams.kstream.Windowed[org.apache.kafka.streams
浏览 2提问于2020-11-18得票数 0
到目前为止，看来我应该把我所有的配置都放到KafkaStreamsConfiguration类型的bean中，以便配置我的kafka-streams应用程序。当我将此配置具体化时，从ProducerConfig和ConsumerConfig在application.properties文件中设置的属性值似乎与它们在spring创建的KafkaStreamsConfiguration中。=latest #this won't show up in KafkaStreamsConf
浏览 1提问于2019-06-13得票数 3
回答已采纳
我正在构建一个spring云微服务，它使用来自kafka主题的数据。在消费者中，我将主题绑定到一个KStream。传入消息不包含时间戳，因为kafka版本低于0.10。否则，当我按键对它们进行分组时，它不会使用"default.timestamp.extractor“(已设置为org.apache.kafka.streams.processor.WallclockTimestampExtractor$StringSerde : streams
浏览 1提问于2019-06-08得票数 0
我是Scala的新手，我正在尝试基于第二个组件字段过滤KStreamString，JsonNode。例如，正在工作的Java代码如下：import org.apache.kafka.streams.kstream.Consumed;import org.apache.
浏览 3提问于2020-03-27得票数 1
回答已采纳
flatMap更改键并从单个消息生成多条消息。
reduce将记录“合并”到包含JSON数组的单个消息中。净效果是将传入的消息(以key作为id1)拆分为多个消息(具有不同的键，例如k1、k2等)。使用join和transform增强每条消息</e
浏览 10提问于2022-10-17得票数 0
我想在外部数据库中做所有的有状态操作，而不是RocksDB，为了做任何需要有状态操作的地方，我编写了自定义处理器，它将进行DB操作，context#forward方法将键值对witten转发到DB到下游消费者
浏览 0提问于2019-09-07得票数 0
1回答
我开发了一个java应用程序，它使用Schema Registry从avro主题读取数据，然后进行简单的转换，并在控制台中打印结果。默认情况下，我对键和值使用GenericAvroSerde类。140) at kafka.tools.ConsoleConsumer$.main(ConsoleConsumer.scala:53)
at <
浏览 0提问于2017-10-05得票数 2
;import org.apache.kafka.streams.KafkaStreamsException in thread "StreamThread-1" java.lang.IllegalArgumentException: Invalid timestamp -1
at org.apache.kafka.clients.producer.Pro
浏览 1提问于2016-07-26得票数 1
云服务器
ICP备案
腾讯会议
云直播
实时音视频
腾讯云开发者
