如何使用Kafka生成连续数字的消息？

文章/答案/技术大牛

发布

1回答

apache-kafka、event-sourcing、event-driven、event-driven-design、event-stream-processing

上下文:开票系统，发送的发票必须有连续的编号。每个发票都有一个惟一的发票编号，为了简单起见，我们假设它们是I1、I2、I3等等。因此，系统中的第一张发票的编号为I1，并且它在每一张下一张发票时递增。然后，每一张发票都是在Kafka主题中产生的。所以，人们总是可以通过这个主题的内容来计算下一张发票的数量，对吗？(主题中的发票数量+1=下一个数字)我们可以将这样的

浏览 18提问于2021-02-04得票数 1

1回答

从kafka broker中查找最后插入的值

apache-kafka、kafka-consumer-api

我遇到了一个API，它每次都会给我一些数字。我必须编写一个Kafka生成器来使用此API输出生成连续的数字流。例如:我点击了API，我得到了450号，然后我必须检查最后在kafka broker中插入的号码(假设是200)。然后在我的生产者中，我必须从201202...450生成数字，并在kafka中作为单独的消息</e

浏览 1提问于2021-07-22得票数 0

1回答

如何在Kafka* Consumer中查看file.txt的输出？*

file-upload、apache-kafka、flume-ng

我正在尝试将Flume与Kafka集成，我想将本地计算机上的一个文件传递给Flume，然后再传递给Kafka。我想在Kafka Consumer中查看我的文件内容。下面是我在Flume中的配置文件： #a1.sinks.k1.type = logger a1.sinks.

浏览 0提问于2015-11-04得票数 1

1回答

测试kafka和flink集成流程

scala、apache-kafka、integration-testing、apache-flink

例如，我想测试卡夫卡/ Flink与FlinkKafkaConsumer011和FlinkKafkaProducer011的集成。这一进程将是：当我说测试时，这是单元/集成测试。谢谢!

浏览 0提问于2018-04-16得票数 3

1回答

spark streaming kafka* :获取主题分区数据时出现未知错误*

apache-spark、apache-kafka、apache-spark-sql、spark-structured-streaming

我正在尝试阅读Spark集群中的Kafka主题，该集群使用结构化流式API与Spark中的Kafka集成 .master("kafka.bootstrap.servers", "<brokerUrl>:9094")

浏览 3提问于2018-09-28得票数 0

1回答

在KSQL中定义双分区属性

apache-kafka、ksqldb

在WITH子句中，我们定义分区是指全新的流(主题)。在GROUP子句中-用于传入消息，以便能够定义到哪个分区发送消息。我们创建了一个包含5个分区的流。让我们想象一下，我们有6个唯一的userid消息。在这种情况下，如何在这5个分区上分发消息？

浏览 1提问于2020-02-23得票数 0

回答已采纳

2回答

无法使用kafka-node向kafka生产者发送消息

apache-kafka、kafka-producer-api、kafka-node

我使用的是Kafka框架提供的默认server.properties/zookeeper.properties文件。module.exports = { kafka_server: 'localhost:9092',node

浏览 14提问于2021-09-14得票数 0

2回答

对于有多个分区的主题，如何保留消息序列？

apache-kafka、kafka-consumer-api、kafka-producer-api

我想知道关于Kafka如何维护消息序列的信息/解释，当消息被写入多个分区的topic时。例如，我有多个消息生成器，每个消息生成器都顺序地生成消息，并使用超过1个分区在Kafka主题上进行编写。在这种情况下，消费者组将如何工作来消费消息。

浏览 0提问于2016-12-01得票数 2

1回答

Akka streams Kafka消费者进程并行

apache-kafka、akka、akka-stream、alpakka、akka-kafka

我正在开发一个使用Akka Kafka连接器的Kafka消费者应用程序。我希望消费者并行处理消息。我应该选择哪个消费群体？？如何在消费者端配置并行度？

浏览 16提问于2020-05-06得票数 0

1回答

使用Snowflake和Kafka连接器的日志和语句

apache-kafka、apache-kafka-connect、snowflake-cloud-data-platform

有没有可能会丢失一些消息，如果我们使用雪花kafka连接器。例如，如果kafka连接器读取消息并在消息写入variant表之前提交偏移量，那么我们将丢失该消息。如果我们使用kafka connect，会出现这种情况吗？如果您有任何示例，也欢迎这些示例，谢谢！

浏览 0提问于2019-12-04得票数 2

2回答

如何在Confluent.Kafka中获得一个主题的所有‘`PartitionTopic`s’的列表？

c#、apache-kafka

我正在使用confluent-kafka-dotnet (Confluent.Kafka)为Kafka生成消息。我想手动管理消息转到哪些分区，以保持某些消息组的顺序。如何获取Kafka主题的PartitionTopic列表？我看过Java中的this solution。但我不明白如何用Confluent.Kafka实现同样<

浏览 113提问于2020-07-14得票数 2

回答已采纳

1回答

关于多节点Akka-Streams瞬时消息的Kafka消费者问题

apache-kafka、akka-stream

我们使用Kafka来存储由集群中的节点生成的消息，并分发到集群中的所有节点，我让它主要使用akka-streams，但有几个问题我必须解决。这是有一些限制的。首先，消息必须由集群中的每个节点使用，但只由一个节点生成。我知道我可以为每个节点分配一个组id，这可能是它的节点ID，这意味着每个节点都会收到消息。已经排序好了。但这里有一

浏览 0提问于2017-12-06得票数 0

3回答

Kafka Consumer没有收到任何来自其生产者的消息

python、apache-spark、apache-kafka、spark-streaming、kafka-python

以下是我为kafka生产者编写的python代码，我不确定这些消息是否可以发布到Kafka代理。因为消费者端没有收到任何消息。当我使用生产者控制台命令对其进行测试时，我的Consumer python程序运行良好。from __future__ import print_function from pyspark import SparkContext from kafka import= KafkaClient("l

浏览 6提问于2016-09-13得票数 0

回答已采纳

1回答

如何使用Esper对来自CSV数据的流执行复杂事件处理？

csv、apache-kafka、complex-event-processing、esper

我使用Kafka生成器连续向Kafka主题发送CSV行，并生成以下形式的CSV行流： temp=23.3, pressure=1015, humidity=63.4, windspeed=7.4....现在，我想使用ESPER在这个流上执行复杂的事件处理，但是Esper使用预定义的POJO类并使用EPL语句来过滤事件。如何在运行时将CSV流转换为事件流，以便Esp

浏览 1提问于2018-07-24得票数 0

1回答

无法使用汇合平台将Avro数据推送到HDFS

apache-kafka、hdfs、avro、apache-kafka-connect、confluent-platform

我有一个系统将Avro数据推入到多个Kafka主题中。在更改主题名称之后，我启动了kafka-connect-hdfs。这一步骤也是成功的。在此之后，我开始将数据推送给Kafka</

浏览 2提问于2017-01-20得票数 1

1回答

消息发送到的Kafka主题的分区id是否与ForeachWriter中的open方法的分区id匹配？

scala、apache-spark、spark-structured-streaming

我有一个要求，我需要创建一个特定于每个分区的文件。我看到消息要发送到的分区与hdfs中的文件并不完全对应。分区在消费者中是如何分配的？

浏览 0提问于2020-04-02得票数 1

2回答

消息不使用kafka* python传递*

python、apache-kafka、kafka-python

我正在尝试用设置一个简单的Kafka应用程序。我一直在尝试让我在网上找到的一些例子起作用，但似乎无法做到。我有一个运行在docker容器中的kafka实例。我测试了shell工具，实例确实工作正常。我能够发送和接收消息。我怀疑生产者消息超时了。下面是代码的两个版本，它们的行为基本相同：from kafka import SimpleProducer, KafkaClient # conne

浏览 21提问于2017-07-09得票数 3

1回答

将实时连续日志数据发送给kafka并进行消费

python、apache-kafka、apache-nifi

我必须分析两个不同应用程序的服务器日志。日志将全天候连续生成。我有生成日志的服务器的凭据。我需要将这些日志数据发送到kafka，并使用python (通过应用regex)从这些日志中提取一些错误信息，并使用我提取的数据准备一个实时仪表板。如何获取持续生成到kafka的日志数据？我需要使用NiFi吗？我对这个场景是新的</

浏览 14提问于2020-03-02得票数 0

1回答

将数据帧从Spark写入Kafka* (org.apache.kafka.common.errors.RecordTooLargeException)时出错*

apache-spark、apache-kafka

我正在使用下面的参数将一个包含30000个条目的Dataframe写入kafka .option("kafka.bootstrap.servers",kafka_brokers) .option("kafka.max.request.si

浏览 0提问于2021-01-30得票数 0

1回答

Apache kafka能否提供异步消息传递服务？

apache-kafka、kafka-consumer-api、kafka-producer-api

我打算使用Spring boot和webSocket以及Kafka实现一个聊天室，但是我需要知道Kafka是否提供了异步消息传递服务

浏览 2提问于2019-07-02得票数 0

点击加载更多