将同一Kafka主题的事件在GCS中分成多个路径
是指将来自Kafka消息队列的事件数据存储到Google Cloud Storage（GCS）中，并按照特定的规则将这些事件数据分散存储到不同的路径中。
Kafka是一种高吞吐量的分布式发布订阅消息系统，常用于大规模数据流处理和实时数据管道。GCS是Google提供的云存储服务，具有高可靠性、可扩展性和安全性。
将同一Kafka主题的事件在GCS中分成多个路径的优势在于：
- 数据隔离：通过将事件数据分散存储到不同的路径中，可以实现数据的逻辑隔离，方便后续的数据处理和管理。
- 并行处理：将事件数据分散存储到多个路径中，可以实现并行处理，提高数据处理的效率和吞吐量。
- 数据归档：按照不同的路径存储事件数据，可以方便地进行数据归档和备份，保证数据的长期保存和可靠性。
- 数据分析：通过将事件数据按照特定规则分散存储，可以方便地进行数据分析和挖掘，提取有价值的信息。
应用场景：
- 实时数据处理：对于需要实时处理大规模数据流的场景，将同一Kafka主题的事件在GCS中分成多个路径可以提高数据处理的效率和并行性。
- 数据归档和备份：将事件数据按照不同的路径存储到GCS中，可以方便地进行数据归档和备份，保证数据的长期保存和可靠性。
- 数据分析和挖掘：通过将事件数据按照特定规则分散存储，可以方便地进行数据分析和挖掘，提取有价值的信息。
推荐的腾讯云相关产品和产品介绍链接地址：
腾讯云提供了一系列与云计算和数据存储相关的产品，以下是一些推荐的产品：
- 腾讯云对象存储（COS）：提供高可靠性、低成本的云存储服务，适用于存储和处理各种类型的数据。产品介绍链接：https://cloud.tencent.com/product/cos
- 腾讯云消息队列（CMQ）：提供高可靠性、高可扩展性的消息队列服务，适用于构建分布式系统和实现异步通信。产品介绍链接：https://cloud.tencent.com/product/cmq
- 腾讯云数据万象（CI）：提供图片和视频处理的云服务，包括图片处理、内容审核、视频转码等功能。产品介绍链接：https://cloud.tencent.com/product/ci
请注意，以上推荐的产品仅为示例，实际选择产品时应根据具体需求进行评估和选择。
相关·内容
1回答
我有一个关于卡夫卡的的问题。我喜欢这个特性，因为在我的例子中，它意味着我可以使用Kafka作为无限的存储空间(例如，使用gcs后端)。但是，让我们假设由于某种原因，Kafka集群被删除，并且Kafka数据丢失。
gcs/s3存储中的数据仍然有用吗？
我的意思是，我可以将旧的日志插入到新的Kafka集群中，或者它现在完全没有用了吗( to级的日志)？
顺便说一句，我知道我可以分析gcs/S3存储中的段并提取数据。但这有点老生常谈，这就是为什么我想看看我是否能找到一个干净的解决方案。
浏览 3提问于2020-10-05得票数 1
我见过很多Apache Beam的例子，你从PubSub读取数据并将其写入GCS bucket，但是有没有使用KafkaIO并将其写入GCS bucket的例子呢？在哪里可以解析消息，并根据消息内容将其放入适当的存储桶中？
例如
<code>A0</code>
我的用例是将数据从Kafka流式传输到GCS存储桶，所以如果有人在GCP中提出了一些更好的方法，也很受欢迎。
谢谢。问候你，阿南特。
浏览 12提问于2020-03-17得票数 1
我正在使用docker与Dockerfile一起安装连接器。我成功地从汇合集线器安装了连接器，但没有安装我自己的jar文件。
以下是我所做的：
转到，在右上角按下下载并单击“jar”将该文件放在与my 运行Dockerfile的文件夹相同的文件夹中：
ENV CONNECT_PLUGIN_PATH="/usr/share/java,/usr/share/confluent-hub-components"
USER root
COPY --chown=appuser:appuser aiven-kafka-connect-gcs-0.7.0.jar /usr/share/con
浏览 5提问于2022-03-05得票数 0
回答已采纳
我对Kafka，zookeeper和Storm都不熟悉。在我们的环境中，我们有一个Kafka代理连接到多个动物园管理员。让生产者将消息发送到特定主题，并将一个代理上的分区发送给多个zookeepers，与将多个代理发送给多个zookeepers相比，是否有优势？
浏览 1提问于2015-09-02得票数 1
我们是Kafka的新手，所以我正在寻找一些高级指导。我们有一个实体的数据(我们可以叫它“订单”)，它本质上是许多不同的实体(我们可以叫一个"Widget“和一个"Gizmo”，但大约有20种不同的实体类型)。 显然，将订单视为单个主题是有好处的，因为所有部件都与一个订单相关。但在设计方面，将这些作为单独的主题(订单、小部件、小工具等)是否更有意义？ Widget和Gizmo之间没有直接的关联--将它们放在一起的好处是处理的顺序等，建议或阅读好的资源将非常有帮助。谢谢!
浏览 13提问于2019-01-11得票数 2
回答已采纳
我有一个弹簧引导项目，我是spring连接到底层的kafka事件中心。
在同一个消费类中，我必须听两个不同的主题。我有两种方法可以这样做。
一个是有两个卡夫卡听众这样的：
@KafkaListener(topics = "topic1")
public void consumeTopic1(String message) throws Exception {
//do something
}
@KafkaListener(topics = "topic2")
public void consumeTopic2(String message)
浏览 1提问于2019-05-08得票数 0
在GCS存储桶上，我从web控制台上传了一个名称为fileName的文件，并创建了一个具有相同fileName名称的目录。因此，GCS存储桶在其根目录中同时包含一个名为fileName的文件和一个同名的目录。
现在，当我尝试通过gsutil命令更新fileName文件时，
gsutil cp localFile gs://bucketname/fileName
其中localFile是我本地计算机上的一个文件，因此gs://bucketname/fileName是dest_url参数，而不是覆盖GCS fileName文件，而是在GCS目录中创建一个名为fileName的新文件，名为fileN
浏览 0提问于2018-06-02得票数 0
我使用的模式注册表与RecordNameStrategy命名策略，所以我有完全不同的avro模式的事件到相同的Kafka主题。 我这样做是因为我想对逻辑上相关的事件进行分组，这些事件可能在同一主题下具有不同的数据结构，以保持这些数据的顺序。 例如：user_created事件和user_mail_confirmed事件可能有不同的模式，但重要的是将它们放在同一个主题分区中，以保证消费者的顺序。 我试图将这些来自单个主题的数据合并到多个路径中的GCS中(每个模式一个路径)。 有没有人知道如果合流的Kafka连接GCS水槽连接器(或任何其他连接器)为我们提供该功能，请？
浏览 9提问于2020-12-17得票数 0
回答已采纳
1回答
我试图让x数的消费者访问kafka中指定的主题，但不使用相同的消息。比如我想要..。
消费者1拾取偏移1消费者2拾取偏移2消费者1拾取偏移3消费者2拾取偏移4
我想让卡夫卡为这两位消费者排队。我注意到了group.id配置，我假设您可以使用同一个组，它会相应地处理它，但它似乎不像我想的那样工作。
这是我使用的代码..。
public void init(){
Properties props = new Properties();
props.put("bootstrap.servers", kafkaUrl);
浏览 2提问于2016-12-06得票数 0
回答已采纳
2回答
我读了很多关于kafka用作活动商店和CQRS的潜在候选者的文章。我想知道，由于kafka中的消息有一个有限的保留时间，那么消息从kafka保留消息的磁盘中删除后，事件将如何重播？
从逻辑上讲，当这些消息从kafka外部(在读取kafka主题的消息之后)存储在数据库(sql/nosql)中时，从事件存储的角度来看，这比kafka更有意义。
相反，如果我的理解是正确的，那么在CQRS中使用kafka的真正用例是什么，尽管kafka的实际意图只是一个高吞吐量的消息系统？
浏览 3提问于2019-11-16得票数 0
现状：
今天，我构建了一个Spark结构化流应用程序，它使用了一个包含JSON消息的Kafka主题。嵌入在Kafka主题的值中包含了有关消息字段的源和模式的一些信息。消息的简化版本如下所示：
{
"source": "Application A",
"schema": [{"col_name": "countryId", "col_type": "Integer"}, {"col_name": "name", "col_type"
浏览 1提问于2018-04-11得票数 3
我希望一个特定的传奇故事的所有消息都发生在同一主题上。 我把我的传奇设置成这样： public void ConfigureServices(IServiceCollection services)
{
services.AddMassTransit(x =>
{
var machine = new MyStateMachine();
var repository = new InMemorySagaRepository<MyState>();
x.UsingAzureServiceBus((ctx,cfg)
浏览 21提问于2020-12-11得票数 0
回答已采纳
我遇到的问题是使用(SCDF)通过环境将数据移动到正确的位置。情况如下：
我有一个生产版的卡夫卡，我只能订阅(只读)。(3节点集群: node 1:9092，Node 2:9092，Node 3:9092)i有一个运行在Kubernetes集群中的SCDF实例，该集群使用RabbitMQ主干(全部在一台服务器上)i有一个节点卡夫卡服务器，用于另一家公司获取数据(ssl受保护)。
在查看SCDF时，我没有看到一种将Kafka定义为源并将其读入处理器(将是自定义的)并写入另一个Kafka实例的方法。
浏览 3提问于2021-06-03得票数 0
我的kafka获得Json格式的包装器有效负载。包装器有效负载如下所示：
{
"format": "wrapper",
"time": 1626814608000,
"events": [
{
"id": "item1",
"type": "product1",
"count": 200
},
{
"id": "item2",
"type": "
浏览 2提问于2022-08-22得票数 0
回答已采纳
背景
有几台机器会产生事件。这些事件被发送到我们的Kafka集群，在这里，每台机器都有自己的主题(app.machine- events . machine -name)。由于顺序在每台机器上都很重要，并且分区大小目前还不是一个问题，所以所有主题都由一个分区组成。因此，N个主题也意味着N个分区，目前。
消费/处理应用程序使用kafka-streams，我们已经给出了StreamsConfig.APPLICATION_ID_CONFIG/"application.id"‘机器-事件-处理器’，它对于每个实例都是一样的，这意味着它们被放入卡夫卡的同一个消费群体中。这个使用者订阅了模
浏览 0提问于2018-10-12得票数 0
回答已采纳
我正在尝试运行一个用Java编写的Spark Kafka Job，以每批产生大约10K条记录到Kafka主题。这是一个spark批处理作业，它在循环中顺序读取100个(总共100万条记录) hdfs零件文件，并批量生成10K条记录的每个零件文件。我正在使用org.apache.kafka.clients.producer.KafkaProducer接口
正在获取以下异常：
org.apache.kafka.common.KafkaException: Failed to construct kafka producer
....
Caused by: org.apache.kafka.comm
浏览 0提问于2020-06-03得票数 0
我将迁移到kafka作为代理和debezium，以便从报告和搜索数据库的所有微服务中获取数据(ETL数据)。
有没有办法配置debezium，使其根据自定义条件(如用户或公司或行/数据的某些关键列/属性)将数据放在不同的主题上。
浏览 0提问于2019-01-24得票数 1
谷歌帮不了我，所以我想问你。我有很多kafka主题，我想将特定主题的消息存储在特定的S3存储桶中。
我是否需要为每个存储桶创建一个S3接收器连接器，或者是否可以在单个S3 json文件中配置所有内容(将存储桶与主题进行映射)？
感谢您的帮助！:)
浏览 0提问于2020-03-19得票数 0
1回答
在不启用SASL或SSL的情况下，我成功地连接了GCS连接器。当我启用SASL和SSL时，连接独立似乎无法与代理进行通信。
问题似乎在于gcs-接收器-许可证管理器。这是我从日志中发现的，但它们对我实际找出问题所在并没有多大帮助.
日志
[2018-12-19 16:29:05,645] INFO [AdminClient clientId=gcs-sink-license-manager] Metadata update failed (org.apache.kafka.clients.admin.internals.AdminMetadataManager:238)
org.apache.
浏览 0提问于2018-12-19得票数 0
回答已采纳
我有一个kafka使用者，它在泛型类中使用avro格式的消息(为了在几个类似的主题中重用它)：
var consumer = new ConsumerBuilder<string, GenericRecord>(conf)
.SetValueDeserializer(new AvroDeserializer<GenericRecord>(schemaRegistry).AsSyncOverAsync())
.Build()
Avro将来自kafka的消息反序列化为这种GenericRecord类型。
Con
浏览 11提问于2022-11-16得票数 1
我正在评估Azure Event Hub与Kafka作为Service Broker的使用。我希望我能够同时创建两个本地应用程序，一个使用Kafka消费消息，另一个使用Azure Event Hub。我已经设置了一个docker容器，它是一个Kafka实例，我正在使用我的Azure帐户设置Azure事件中心(据我所知，没有其他方法可以为Azure事件中心创建本地/开发实例)。
有没有人有任何关于这两者的信息，在比较它们的功能时可能会有用？
浏览 35提问于2020-09-18得票数 6
在Kafka文档中，我无法理解划分一个主题的确切需要是什么？如果它只是一个没有分区的主题(或者只有一个分区)，问题是什么？
浏览 8提问于2022-12-02得票数 0
2回答
关于卡夫卡，你有没有遇到过类似的问题？我得到这个错误：Too many open files。我也不知道原因。以下是一些日志：
[2018-08-27 10:07:26,268] ERROR Error while deleting the clean shutdown file in dir /home/weihu/kafka/kafka/logs (kafka.server.LogD)
java.nio.file.FileSystemException: /home/weihu/kafka/kafka/logs/BC_20180821_1_LOCATION-87/leader-epoch-
浏览 102提问于2018-08-27得票数 2
回答已采纳
1回答
我是Kafa和数据摄取的新手。我知道Kafka是容错的，因为它将数据冗余地保存在多个节点上。然而，我不明白的是，我们如何在源/生产者端实现容错。例如，如果我使用netcat作为资源，如下面的示例所示。
nc -l [some_port] | ./bin/kafka-console-producer --broker-list [kafka_server]:9092 --topic [my_topic]
如果执行netcat的节点宕机，生产者将无法推送消息。我在想，是否有一种机制可以让Kafka自己拉取输入，例如，如果在一个节点上netcat出现故障，另一个节点可以接管并开始使用netcat推送
浏览 6提问于2020-05-21得票数 2
我有一个Spark结构化流媒体应用(v2.3.2)，它需要从一些Kafka主题读取，做一些相对简单的处理(主要是聚合和一些连接)，并将结果发布到其他一些Kafka主题。因此，在同一个应用程序中处理多个流。 我想知道从资源的角度(内存、执行器、线程、Kafka侦听器等)它是否会有所不同。如果我只设置一个直接readStream，它订阅多个主题，然后使用selects拆分流，而不是每个主题1个readStream。 就像这样 df = spark.readStream.format("kafka").option("subscribe", "t1,t2,t
浏览 28提问于2019-05-01得票数 11
回答已采纳
不同的事件类型由主题/队列和消息头的某些组合来区分。在Kafka中，记录流仍然作为主题()分开。在阅读Azure事件中心文档时，我没有看到任何这样的想法。有分区，但根据文档，这些分区与使用者并行性()有关。您是否认为在Azure事件集线器中有一个名称空间，其中包含每个“主题”的事件中心？
浏览 2提问于2017-02-15得票数 3
回答已采纳
我读过几篇关于使用Kafka和Kafka流(与状态存储一起)作为事件存储实现的文章。
实施理念如下：
在kafka主题中存储实体更改(事件)
使用Kafka流和状态存储(默认情况下使用RethinkDB)更新和缓存实体快照
每当执行新命令时，从商店获取实体，执行对其的操作，并继续执行步骤1
此工作流的问题是，State正在异步更新(步骤2)，当处理一个新命令时，检索的实体快照可能会失效(因为它没有使用以前命令中的事件进行更新)。
我的理解正确吗？有什么简单的方法来处理卡夫卡这样的案子吗？
浏览 20提问于2019-12-05得票数 0
1回答
卡夫卡对库伯奈特连接问题的影响
、、、、
我在连接Fluentd到Kafka以获得我正在处理的集中式日志记录PoC时遇到了问题。
我目前正在使用以下配置：
米尼库
弗伦特
fluent/fluentd-kubernetes-daemonset:v1.14.3-debian-kafka2-1.0 (码头)
配置:我在yaml中设置了FLUENT_KAFKA2_BROKERS=<INTERNAL KAFKA BOOTSTRAP IP>:9092和FLUENT_KAFKA2_DEFAULT_TOPIC=logs env，用于流畅的守护进程。
卡夫卡
我有点期待看到在Kafka消费者中出
浏览 3提问于2022-02-10得票数 1
1回答
我是NATS Jetstream的新手，我一直在阅读他们的官方文档(https://docs.nats.io/jetstream/jetstream)，以理解它的概念，并将其与Kafka进行比较。我的一个主要用例是解决基于特定id的消息/事件排序(就像Kafka世界中的partition key )。 例如，有几个Order实体的更新事件即将到来，而我的系统需要以相同的顺序使用特定Order的事件。在这种情况下，当发布到Kafka主题时，我将使用order-id作为分区键。如何在Jetstream中完成此操作？ 我在Jetstream中遇到了去重键(Nats-Msg-Id)，但我认为这个功能
浏览 216提问于2021-08-30得票数 5
我有一个要求建立聚合的数据，我们收到我们的Apache Kafka… 我有点不知道该走哪条技术路线…… 似乎人们看到的是标准的方式，一群Apache Kafka <-> Apache Spark <-> Solr Bitnami Data Platform 我找不到具体的例子说明这是如何工作的，但我也在问自己，是否有任何解决方案 Apache Kafka <-> Kafka Connect Solr <-> Solr 不会这样做，因为solr也支持聚合... Solr Aggregation 但我看到了一些代码片段，它们将数据聚合在Spark中
浏览 34提问于2021-11-11得票数 0
我尝试在分布式模式下用一个工作者将一个Kafka Connnect实例的信息发送到一个特定的主题，我在启动实例时使用的"archive.properties“文件中有主题名称。
但是，当我发送五个或更多实例时，我看到消息合并到所有主题中。
我想的“解决方案”是建立一个映射来存储ID和主题之间的关系，但它并不起作用
有没有特定的Kafka connect实现来做这件事？
谢谢。
浏览 10提问于2017-01-20得票数 0
1回答
我制作了一个kafka管道，将sqlserver表复制到s3
在接收器期间，我尝试用regex路由器函数转换主题名称删除前缀：
"transforms":"dropPrefix",
"transforms.dropPrefix.type":"org.apache.kafka.connect.transforms.RegexRouter",
"transforms.dropPrefix.regex":"SQLSERVER-TEST-(.*)",
&#
浏览 0提问于2018-11-15得票数 0
对于kafka服务器内部的分流，有哪些不同的选择？
1) Split the traffic into different topics.
2) Split the traffic into different partitions, after being assigned a topic.
3) Split the traffic into different slots (am not this exist).
谢谢
浏览 2提问于2018-05-07得票数 1
在Flume中，当使用Kafka频道时，是否有办法影响事件被发送到哪个分区？
使用Kafka 接收器，key FlumeEvent头显然用于选择一个分区，但我找不到任何关于使用Kafka 通道的分区的文档。
浏览 3提问于2016-02-02得票数 0
回答已采纳
我们使用spring库和@Listener注释编写的Kafka用户似乎在随机跳过消息。这并不是每次都发生，而是足够的次数，让我们调查。似乎发生的时间更多的时候，事件发布后的时间间隔，即消费者没有任何信息处理一段时间。
我们正在使用AWS MSK (AWS管理的kafka)。
主题具有2Ack模式的复制因子，所有自动偏移复位是“最早”的。
我们同时尝试使用kafka控制台消费者，这似乎是在接收信息。
以前有没有人遇到过这样的问题？我们可以调整哪些配置？
浏览 6提问于2022-09-06得票数 0
我有一个kafka容器开始使用下面的代码
docker run --detach --name kafka -p 2181:2181 -p 9092:9092 --env ADVERTISED_HOST=192.168.1.89 --env ADVERVTISED_PORT=9092 --env AUTO.CREATE.TOPICS.ENABLE spotify/kafka
我可以使用docker logs kafka来查看它的启动。
然后，我创建了一个简单的groovy脚本客户端生成器来编写一些条目，但是这会导致错误
> Sending metadata request {topics
浏览 3提问于2017-01-29得票数 2
我是KAFKA协议世界的新手，我想问你一些与我的项目相关的重要信息。
我正在使用AVRO文件来生产和消费消息，我想知道是否可以将相同的avro文件用于多个主题，例如，通过在生产者中使用不同的"name“属性，以及在使用者中使用特定的"name”属性。
非常感谢。斯特凡诺
浏览 0提问于2018-07-20得票数 0
2回答
为了分离我的数据，基于一个键:我应该在同一主题中使用多个主题还是多个分区？我是根据服务器的开销、计算、数据存储和负载来计算的。
浏览 0提问于2015-09-15得票数 0
我最近开始学习apache kafka。在得到一些初步的想法后，我对kafka在应用程序架构中的位置感到困惑。
我的意思是，我们如何将它引入到以前只有三层的简单应用程序中。前端(安卓应用)，后端(Node.js)，MySQL。用户可以从应用程序创建帖子，应用程序将调用node.js上的端点，并将帖子存储在数据库中。
在这个简单，直截了当的架构中，kafka还会添加什么额外的东西？它将如何集成？
浏览 2提问于2021-10-27得票数 0
1回答
我需要一个建议，什么是正确的卡夫卡结构使用我的项目和原因。
我的项目是为投资机器人管理创造一个平台。非常高水平-你可以编码几个投资策略，并将它们上传到平台上，它们将实时执行，提供分析和性能的实时信息。这些策略是从4个数据流中输入信息的。当他们阅读四个不同的卡夫卡主题时，这些数据被传递给策略。这个kafka主题直接从交换websocket接收信息。在任何给定的时间，平台中都有一个动态的机器人数量。
我所做的是:使用图像卡夫卡-wurmeister和动物园，初始化卡夫卡初始化所有的卡卡主题，我将需要事先。我将所需的数据推送给Kafka，方法是将所有信息生成到以下主题：
payloads = [
浏览 3提问于2020-05-10得票数 0
回答已采纳
我想获得帮助的初始设置与apache storm和kafka设置。
我可以向storm集群提交拓扑，但在storm ui中出现以下错误。
Unable to get offset lags for kafka. Reason: java.lang.IllegalArgumentException: zk-node '/kafka-cluster-1/brokers/topics/myfirsttopic/aadb3eb4-2224-4c18-b8ad-6959a1c9f607' dose not exists. at org.apache.storm.kafka.monitor
浏览 34提问于2018-04-21得票数 0
回答已采纳
1回答
更新Kafka事件日志
、、、
我正在使用卡夫卡作为管道来存储分析数据，然后再将其发送到S3，并最终发送到Redshift。我正在考虑用Kafka存储数据的最佳架构，这样就可以轻松地将数据存储到数据仓库中。
问题是，我从三个单独的页面事件中获取数据：
当页面被请求时。
加载页面时
当页面卸载时
这些事件发生在不同的时间(通常在几秒钟之内，但距离对方最多几分钟/小时)。
我希望最终在我的数据仓库中存储一个关于网页视图的单一事件。例如，一个日志条目如下：
pageid=abcd-123456-abcde, site='yahoo.com' created='2015-03-09 15:
浏览 1提问于2015-03-09得票数 0
回答已采纳
我有一个部署，我们使用kafka从服务发送消息。但我们需要卡夫卡大师在所有地区。因此，一旦消息被推送到一个数据中心，它就应该被同步到其他数据中心。当它再次在其他数据中心完成时，应该同步回同步。镜像制造者可以提供从1到其他同步，但我如何实现双向同步？
浏览 1提问于2016-08-18得票数 4
回答已采纳
我有一个包含多层文件夹的存储桶，其中包含1000个文件(而且还在不断增长)。我想知道GCS是否有办法导出整个存储桶的csv报告？
例如，亚马逊网络服务S3通过其存储库存提供该服务。
GCS storage api有一个/list端点，可以在过渡期间提供帮助，但是当单个存储桶中有1000-1亿个对象时，它似乎不是一个可伸缩的选项？
浏览 0提问于2019-08-16得票数 2
我有一个Lambda函数，它有多个MSK触发器配置--每个配置针对不同的主题。
如果Lambda的输入()可以包含多个不同的主题，则在正式文档中找不到任何信息。
正式文档为具有单个mytopic主题的输入提供了以下示例：
{
"eventSource":"aws:kafka",
"eventSourceArn":"arn:aws:kafka:sa-east-1:123456789012:cluster/vpc-2priv-2pub/751d2973-a626-431c-9d4e-d7975eb44dd7-2",
浏览 5提问于2022-02-13得票数 2
3回答
| 的所有都显示ProducerRecord的键/值对不仅具有相同的类型(所有示例都显示为<String,String>)，而且具有相同的值。例如：
producer.send(new ProducerRecord<String, String>("someTopic", Integer.toString(i), Integer.toString(i)));
但在Kafka文档中，我似乎找不到在哪里解释了键/值的概念(以及它的基本用途/工具)。在传统的消息传递(ActiveMQ、RabbitMQ等)中我总是在特定的主题/队列/交换上触发一条消息。但Ka
浏览 20提问于2016-11-30得票数 116
回答已采纳
1回答
卡夫卡消费者没有消费
、、、
我正在尝试创建一个带有10个分区的Kafka主题的应用程序。我没有在application.properties中定义消费者的分区数，因为应用程序将部署在OpenShift中，我希望OpenShift完成负载平衡。问题是，当我启动应用程序时，它没有消耗任何东西。代码非常简单，我真的不知道哪里出了问题：
这是使用者，日志没有显示任何内容：
@Incoming("plac-fate")
@Transactional
public void consume(NFeDistSVBAPayload payload) throws PlacException {
浏览 6提问于2022-10-11得票数 1
回答已采纳
要在Kafka中创建主题，命令为
bin/kafka-create-topic.sh --zookeeper localhost:2181 --replica 1 --partition 1 --topic test
这很简单。但是我如何创建两个不同的主题，两个主题在kafka中都有相同的消息集？
浏览 0提问于2015-08-04得票数 0
我有一个要求，在Kafka中，我们将把一堆不同提供商的财务价格反馈到一个单独的“市场数据”主题中。问题是，这些提供商中的一些人有细粒度的许可要求，甚至到个人证券。在不为每个不同的许可粒度创建不同的主题的情况下，有没有一种方法可以防止Kafka客户端根据消息中的内容接收信息？即Kafka可以做细粒度的主题内许可吗？ 如果不是，那么这个问题的标准解决方案是什么？
浏览 10提问于2019-06-30得票数 2
回答已采纳
我仍然不能清楚地理解并行性，比方说，我们有一个flink集群，它有足够的插槽。在我们的flink工作中，我们从3个不同的kafka集群中使用3个kafka主题，每个主题有10个分区。
如果我们想尽快使用该消息。并行性应该是什么？
如果我们将并行性设置为10，那么就会使用10个槽，这意味着，根据我的理解，10个线程对吗？如果这10个线程“连接”到topic_1，那么就没有线程读取topic_2和topic_3。
如果我们将并行性设置为30，那么就有30个线程，但这30个线程是否足够聪明，它们中的10个会转到topic_1，10个会转到topic_2，其余的10个会转到topic_3？
浏览 0提问于2019-03-13得票数 12
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
