Ignite Source Kafka Connector是否提供数据的主题分区？
Ignite Source Kafka Connector是Apache Ignite的一个组件，用于从Apache Kafka读取数据并将其加载到Ignite内存中。该连接器提供了许多功能，包括数据的主题分区。
数据的主题分区是指将数据根据其键分散存储在不同的分区中，从而实现数据的分布式存储和处理。Ignite Source Kafka Connector允许根据Kafka消息的键对数据进行分区，以便在Ignite集群中实现高效的并行处理和查询。
Ignite Source Kafka Connector的主题分区功能可以通过配置文件进行设置。在配置文件中，可以指定分区函数和分区数目。分区函数用于确定数据分配到哪个分区，而分区数目定义了Ignite集群中的分区数。通过灵活的配置，可以根据具体的业务需求对数据进行精确的分区。
Ignite Source Kafka Connector的主题分区功能具有以下优势：
- 分布式处理：主题分区使得数据可以在Ignite集群中并行处理，提高数据处理速度和吞吐量。
- 负载均衡：通过将数据均匀分布在多个分区中，主题分区可以实现负载均衡，避免某些分区过载。
- 高可用性：Ignite集群中的数据可以复制到不同的分区，提供数据的冗余备份，确保系统的高可用性。
- 数据本地性：主题分区可以确保具有相同键的数据被存储在同一分区中，从而提高数据的本地性和查询性能。
Ignite Source Kafka Connector的主题分区功能适用于以下场景：
- 大规模数据处理：当需要处理大量数据并实现分布式处理时，主题分区可以提高处理效率。
- 实时数据分析：通过将数据分区存储，可以实现实时数据分析和查询。
- 高性能计算：主题分区可以为高性能计算提供分布式数据存储和处理能力。
推荐的腾讯云相关产品是Tencent Distributed Data Store (TDS)，它是腾讯云提供的分布式存储服务。TDS支持分布式数据存储和处理，可以与Ignite Source Kafka Connector配合使用，实现数据的主题分区和高性能计算。您可以通过以下链接了解更多关于TDS的信息：https://cloud.tencent.com/product/TDS
相关·内容
1回答
我开始运行一个自定义的kafka连接插件，将kafka中的数据转储到ElasticSearch中，然后我的任务就会周期性地无缘无故地死亡(几个小时到一天后)
我在连接状态中看到了以下内容：
curl -s http://localhost:8083/connectors/my-custom-sink/status | python -m json.tool
{
"connector": {
"state": "RUNNING",
"worker_id": "MASKED_IP1:80
浏览 4提问于2017-07-31得票数 4
回答已采纳
我有一个在Confluent Cloud中运行的集群，并且能够使用其他应用程序生成和使用数据。然而，当我尝试连接Snowflake Kafka连接器时，我收到以下错误：
[2019-10-15 22:12:08,979] INFO Creating connector source-snowflake of type com.snowflake.kafka.connector.SnowflakeSinkConnector (org.apache.kafka.connect.runtime.Worker)
[2019-10-15 22:12:08,983] INFO Instantiated c
浏览 0提问于2019-10-16得票数 2
1回答
kafka点火水槽连接器
、、、
我正在尝试将数据从Kafka topic复制到apache ignite集群。我已经在插件中添加了ignite-core.jar以及所有其他所需的文件，并根据日志将其加载到连接器中。尽管如此，我仍然面临着这个错误。任何建议都会很有帮助。谢谢。
INFO Instantiated connector string-ignite-connector with version 6.1.1-ccs of type class org.apache.ignite.stream.kafka.connect.IgniteSinkConnector (org.apache.kafka.connect.
浏览 30提问于2021-05-15得票数 1
这是关于卡夫卡MirrorMaker工具。
我在两台机器上配置了卡夫卡。
来源：
目的地: vm仅位于源端
卡夫卡在来源和目的地都是同样版本的卡夫卡卡夫卡_2.11-0.9.0.0
在源和目的地，各自的动物园管理员和卡夫卡服务器正在运行。
使用MirrorMaker工具，我想从源到目的地复制/制作主题镜像。
下面是我使用的命令：
./bin/kafka-run-class.sh kafka.tools.MirrorMaker --consumer.config ./config/mirror_consumer.properties --producer.config .
浏览 7提问于2016-05-06得票数 2
1回答
在使用ErrorHandlingDeserializer结合Avro处理错误时，我无法发布到Dlq主题。下面是发布时的错误。
主题TOPIC_DLT在60000 ms后不存在于元数据中。错误KafkaConsumerDestination{consumerDestinationName='TOPIC'，partitions=6，dlqName=‘TOPIC_DLT’}。容器-0-C-1 o.s.i.h.LoggingHandler:250 - org.springframework.messaging.MessageHandlingException:错误发生在消息处理程序or
浏览 3提问于2022-03-02得票数 0
回答已采纳
1回答
我试图通过kafka连接器将467个主题上的大约2797个分区推送到snowflake，而(2个任务) kafka连接器使用：
[2020-03-08 10:13:10,884] ERROR WorkerSinkTask{id=snowflake-1} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask)
java.lang.OutOfMemoryError: unable to create native thread: possibly out
浏览 31提问于2020-03-08得票数 1
我想将我的Kafka主题数据复制到ignite节点。但情况是Kafka服务器运行在AWS实例上，而Ignite运行在我的机器上。那么，我提供给connector.properties的XML文件的内容应该是什么呢？提前谢谢。
浏览 2提问于2021-05-14得票数 1
我使用的是与kafka 2.0.0汇合的hdfs接收器连接器5.0.0，我需要使用ExtractTopic转换()。我的连接器工作得很好，但是当我添加这个转换时，我得到了NullPointerException，即使在只有2个属性的简单数据样本上也是如此。
ERROR Task hive-table-test-0 threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerSinkTask:482)
java.lang.NullPointerException
at io.c
浏览 0提问于2019-01-25得票数 0
回答已采纳
我正在创建一个CDC Postgres.source --> Kafka / Debezium --> Postgres.sink。 源主题创建得很好，但只有在表有数据的情况下才能创建。有没有办法在没有数据的情况下实例化所有的主题？因为我想用maven命令下载所有AVRO模式： mvn schema-registry:download 我的配置： {
"name": "postgres-source",
"config": {
"connector.class":"io.debezium.co
浏览 20提问于2021-07-30得票数 0
2回答
我正在使用kafka-connect + Debezium捕获我的MySQL集群中的数据更改。目前，我们使用Debezium的table.whitelist来摄取MySQL集群中的一小部分表。每当我向table.whitelist配置添加一个新表时，Debezium将创建一个只使用replication_factor = 2和partition=1的新的kafka主题。我的Kafka集群共有4名经纪人(1名领导人和3名追随者)，在我的/opt/kafka-connect/worker.properties文件中，我设置了以下内容：
offset.storage.replication.fa
浏览 4提问于2021-05-25得票数 0
我使用Kafka Connect添加了JDBC Source Connector，Kafka Connect从MySQL表中获取数据。
连接器添加成功，数据在其中实时流动。
但相应的主题并不包含所有先前的数据。
我尝试过重启Kafka Connect。
下面是我的JDBC源连接器配置：
{
"name": "kb_yp_loan",
"connector.class": "io.aiven.connect.jdbc.JdbcSourceConnector",
"tasks.max": "1",
&#
浏览 4提问于2019-11-04得票数 0
我已经从MySQL创建了一个示例管道轮询数据，并将其写入HDFS(也包括hive)。
由于我的需求，我需要为每个db表创建Source+Connector对。接下来，我已经发布了我的源代码和Sink连接器的配置设置。
我可以看到一个主题是用一个分区和复制因子1创建的。
主题创建应该是自动的，这意味着我不能在创建Source+Sink对之前手动创建主题。
我的问题：
1)在创建源连接器时是否有方法配置分区数和复制因子？
2)如果可以创建多个分区，源连接器使用何种分区策略？
3)源连接器和水槽连接器的正确数量应该是多少？
源连接器：
{
"connector.class": &
浏览 0提问于2019-02-13得票数 2
回答已采纳
I am trying to implement a prototype for implementing messaging system using Spring Cloud Stream. I selected Apache Kafka as binder. I created a topic with 2 partitions for scalability. Then I tried to send different messages to different partitions using following rest api method.
我为两个分区设置了两个不同的
浏览 7提问于2022-06-30得票数 0
回答已采纳
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
我试图发布数据从卡夫卡主题到弹性搜索使用卡夫卡连接。下面是我的配置。
{
"name": "elasticsearch_sink_19",
"config": {
"connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
"tasks.max": "1",
"topics": "k-connect-status"
浏览 2提问于2020-06-10得票数 0
回答已采纳
我的问题是Apache Ignite现在有了putAsync()方法，它实际上是如何将我的数据排入队列的？它会进入内存吗？它是否会写入磁盘上的日志？一旦我的数据完成后，它是如何从未来进入数据库的？ 例如，Kafka，当你想发送数据时，你需要调用producer.send()，它是异步的，我相信Kafka会将你的请求放入一个缓冲区，后台线程会通过网络刷新该缓冲区。如果该缓冲区超过限制，则它将抛出异常。 因此，对于Kafka，我知道有另一个线程在后台刷新数据&根据缓冲区大小，我可以进行多少异步调用是有限制的。 我正在为Ignite寻找类似的解释！
浏览 32提问于2021-01-14得票数 0
我用debezium插件运行zookeeper，kafka和kafka连接。我看不到任何错误。我已经把一个数据库和几个表列入了白名单。在插入或更新表时，不会创建主题；/tmp/kafka-logs中的主题也不会有数据目录。我刚刚尝试为其中一个表创建一个主题(一种只有一列的虚拟表't‘)，但是在t中插入一行之后，该主题的日志中没有数据...日志的屏幕截图如下 只有一个主题(我创建的那个主题，看看如果我自己创建它，它是否会被填充；我没有想到它会被填充，它也不是)： vagrant@coton:~$ ./kafka/kafka_2.13-2.8.0/bin/kafka-topics.sh
浏览 33提问于2021-07-25得票数 0
我运行了一个MQTT代理和一个Kafka代理，我使用了kafka-connector：，配置如下：
name=Mqtt-Source
connector.class=com.datamountaineer.streamreactor.connect.mqtt.source.MqttSourceConnector
tasks.max=1
connect.mqtt.kcql=INSERT INTO test SELECT * FROM + WITHCONVERTER=`com.datamountaineer.streamreactor.connect.converters.source.JsonS
浏览 0提问于2018-11-19得票数 1
设置:在MS上启用了，并且使用(源代码)将事件发送给。此外，在Kafka中，不止一个表CDC事件被路由到一个单一的主题。
问题:由于主题中有多个表数据，所以我希望在CDC数据中有表名和数据库名。
我在MySQL CDC中获取表名和数据库名，而不是在MySQL中。
下面是Server的Debezium源连接器
curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d '{
"name&
浏览 1提问于2019-05-01得票数 1
回答已采纳
我使用sql-server kafka-connect和debezium来监视sql server数据库，但是当我发布并运行我的工作程序时，主题已经用名称{server_name}.{schema_name}.{table_name}和默认的partitions=1 & replication=1创建了，即使我在"bootstrap.servers“参数中指定了两个代理。如何手动更改表主题的分区和复制，或者在worker/connector配置中预先指定？此外，不同的主题可能有不同的分区和复制。
浏览 0提问于2019-12-30得票数 2
默认情况下，多个Kafka消费者可以从同一个topic的同一分区中读取数据吗？默认情况下，我的意思是因为group.id不是强制性的，所以我想知道如果我在没有指定任何group.id的情况下产生多个kafka使用者，并为它们提供相同的主题和分区名称，它们是否能够从相同的分区中读取？我知道，如果我为每个Kafka消费者提供不同的组名，那么所有消费者都可以从同一个分区中读取数据。
浏览 4提问于2017-03-31得票数 1
2回答
我们正在考虑使用kafka进行分布式开发，但也希望将其用作数据库。具体情况:我们用kafka写“交易”主题，并希望依赖它来存储所有的交易。问题是:在这个设计中是否需要一个恢复计划，卡夫卡会不会因为崩溃、磁盘故障而丢失数据？或者卡夫卡有自己的恢复机制，所以用户不需要自己的恢复计划？
浏览 0提问于2018-04-06得票数 3
回答已采纳
1回答
我的目标是建立一个高吞吐量集群，使用Kafka作为源，使用Flink作为流处理引擎。这就是我所做的。
我已经在主服务器和工作人员上设置了一个2节点集群，配置如下。
主flink-flink.Master
jobmanager.rpc.address: <MASTER_IP_ADDR> #localhost
jobmanager.rpc.port: 6123
jobmanager.heap.mb: 256
taskmanager.heap.mb: 512
taskmanager.numberOfTaskSlots: 50
parallelism.default: 100
W
浏览 7提问于2015-08-28得票数 9
回答已采纳
1回答
我正在尝试使用JMX收集kafka主题指标，这样我就可以将它们打包到java中的对象中。Kafka Docs显示MBean需要一个client-id，但是我们的系统中没有设置任何client-id。 让我相信可以在没有客户端id的情况下获得主题级别指标的原因是，我们的Kafka Manager服务实例能够提取主题指标。我找到了他们使用"kafka.server:name=MessagesInPerSec,topic=topic,type=BrokerTopicMetrics"的MBean，并尝试使用它，但没有结果。 我当前的代码是 try {
JMXServiceURL
浏览 15提问于2019-05-07得票数 1
回答已采纳
我使用的是带有分片副本集集群的MongoDB atlas，以及文档中描述的。
下面是我当前的配置(运行独立设置)：
name=dev-mongodb
connector.class=io.debezium.connector.mongodb.MongoDbConnector
tasks.max=4
mongodb.hosts=<some-url>.mongodb.net:27017
mongodb.name=mongodb
mongodb.user=<admin_user>
mongodb.password=<admin_user_pw>
database.
浏览 0提问于2021-04-24得票数 0
当Kafka消费者从其分配的分区中读取数据时，消费者抓取器线程是否使用任何特定的逻辑来从分区中获取数据？例如，抓取器线程是否会做出任何逻辑/努力来均匀/统一地从分配的分区中读取数据？它是否会从最滞后的分区中获取更多的记录？或者这只是一个简单的类似循环的逻辑？
有没有详细的文档来说明消费者的使用逻辑？
谢谢。
浏览 4提问于2021-07-23得票数 0
1回答
我使用了debezium.io 提供的教程，并将其调整为使用MySQL数据库而不是MySQL，但是当我在数据库中进行更改时，观察者不会显示任何事件或活动。以下是我所采取的步骤：
我运行了动物园管理员码头命令：
docker run -it --rm --name zookeeper -p 2181:2181 -p 2888:2888 -p 3888:3888 debezium/zookeeper:1.1
然后我运行了kafka停靠命令：
docker run -it --rm --name kafka -p 9092:9092 --link zookeeper:zookeeper de
浏览 4提问于2020-06-12得票数 2
回答已采纳
我有一个用例，我想先建立一个Kafka集群，一开始我有一个Kafka代理(A)和一个动物园管理员节点。因此，以下是我的质询：
在集群中添加一个新的Kafka Broker(B)。代理A上的所有数据都会自动分发吗？如果不是的话，我需要做的是分发数据。
不让我们想当然地假设这件事！我的数据分布在两个经纪人身上。现在，由于一些维护问题，我想要删除服务器B。
- How to transfer the data of Broker B to the already existing broker A or to a new Broker C.
如何在运行时增加代理的复制因子？
浏览 0提问于2018-12-07得票数 0
1回答
我编写了一个Akka应用程序，它从Kafka获取输入，然后与共享的演员处理数据并输出给Kafka。
但在某些情况下，切分区域无法处理负载，我得到：
您可能应该实现流控制，以避免淹没远程连接。
我如何在这个链条/流程中实现背压？
卡夫卡消费者->分享演员->卡夫卡制片人
代码中的一些片段：
ReactiveKafka kafka = new ReactiveKafka();
Subscriber subscriber = kafka.publish(pp, system);
ActorRef kafkaWriterActor = (ActorRef) Source.act
浏览 4提问于2017-04-10得票数 0
我正在尝试用MirrorMaker 2.0复制卡夫卡集群。我使用以下mm2.属性：
# mm2.properties
clusters=source, dest
source.bootstrap.servers=kafka-1:9092,kafka-2:19092,kafka-3:29092
dest.bootstrap.servers=kafka-4:39092,kafka-5:49092,kafka-6:59092
source->dest.enabled=true
我可以看到，当MirrorMaker启动时，它会创建自己的主题。例如，它在源集群上创建：mm2-configs.d
浏览 5提问于2020-01-23得票数 2
1回答
我试图使用Kafka连接来使用独立模式编写数据。我正在编写数据的主题是有多个分区。但是，数据只被写入其中一个分区。当我启动多个使用者控制台时，数据只会打印到其中一个。另一个使用者控制台只有在第一个用户控制台关闭后才能获得任何数据。我无法确定需要在配置文件中进行哪些更改才能使其写入多个分区。
这是standalone.properties
bootstrap.servers=localhost:9092
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connec
浏览 1提问于2017-06-26得票数 0
回答已采纳
1回答
我有两个微型服务，一个生产者和一个消费者。制片人每两秒为卡夫卡主题写一个数字的增量。使用者有两个实例正在运行，使用这些增量。我注意到了一些我想解决的奇怪问题：
当还没有消费者的时候，制片人开始制作这些信息就被保存在卡夫卡里。当消费者在线时，它不会处理生产者已经生成的已经存在的消息，而是开始使用现在传入的消息。消费者如何处理所有未消耗的消息？
当有两个消费者时，我希望两个消费者的消费是平等的。现在只有一个消费者承担了所有的负担，而另一个只是坐在那里。我如何将负载分散到消费者的数量上呢？
看起来卡夫卡保存了所有正在制作的唱片，即使它已经被消费者消费了。有什么办法可以防止这种情况发生吗？
浏览 1提问于2020-07-12得票数 0
嗨，我正在使用debezium捕获Mongo中的更改并将它们推入mysql --我使用下面的链接将end替换为mysql数据库，但我无法这样做。
这是我修改过的jdbc-sink.json，在这里我使用mysql url进行连接。
{
"name" : "jdbc-sink",
"config" : {
"connector.class":"io.confluent.connect.jdbc.JdbcSinkConnector",
"tasks.max"
浏览 0提问于2019-11-06得票数 0
1回答
我想用kafka实现一个队列机制。但是找不到任何地方，如果可以只偷看为任何主题创建的队列中的数据，而不进入其中。
我想从队列中读取数据，并根据不同的条件想要删除现有的消息或将另一条消息添加到此队列中。另外，是否可以使用来自不同机器的单个kafka服务器。
我参考了tutorialspoint以了解更多关于它的信息。
提前谢谢。任何线索都将不胜感激。
浏览 0提问于2018-07-26得票数 0
目前，我们使用Flink FsStateBackend检查点并设置fileStateSizeThreshold，将写入HDFS上avro/json文件的数据大小限制在128 on。在检查点操作延迟一定时间后也关闭文件。
由于我们在新项目中没有使用高级Flink特性，所以我们希望使用Kafka Connect HDFS Connector中的Kafka流直接将消息写入hdfs (而不是旋转Flink)。
但是，我无法找到限制卡夫卡连接器中hdfs文件的文件大小的选项，除了flush.size，它似乎限制了记录的#。
如果连接器上没有设置，人们如何以另一种方式管理hdfs上流数据中的文件大小？
浏览 0提问于2018-08-08得票数 1
回答已采纳
2回答
Spark流分区
、、
我们正在使用Kafka和Spark streaming来处理贸易数据。我们从卡夫卡接收avro格式的数据[key，byte[]]。我们对数据进行反序列化，并将其进一步发送以进行处理。我们在spark流媒体应用中使用了DStreams。我们需要根据接收到的avro记录中的关键字对数据进行分区。因此，每当我们以流的形式从kafka接收数据时，它都应该将记录发送到指定的执行器。
我们从Kafka那里收到了10种不同类型的密钥。所以所有包含key1的记录都应该放到Node1中，key2应该放到Node2中，等等。
作为接收到的流数据，我们映射到RDD，而不是pairRDD。
请告诉我们是否可以根据从k
浏览 1提问于2018-07-27得票数 0
我试图创建一个卡夫卡连接连接器，从一个AVRO主题到一个文件。
然后使用kafka-connect将该文件还原到另一个主题。
水槽工作正常，我可以看到接收器文件正在增长，并读取数据。但是当我试图恢复到一个新的主题时，新的主题仍然没有数据。
我没有错误，我已经重置了偏移量，我创建了一个新的Kafka连接并尝试恢复，我创建了一个完整的新Kafka集群，并且始终是相同的，源连接器上没有错误，但是主题是空的。
在这里，源连接器配置的输出：
{
"name": "restored-exchange-rate-log",
"config": {
浏览 1提问于2019-02-08得票数 1
3回答
我正在尝试将一条消息路由到Kafka中的死信主题，以防在处理相应消息时出现任何失败。我已经为该功能设置了SeektoCurrentErrorHandler和DeadLetterPublishingRecoverer。
使用者在执行此操作时抛出下列异常：
2020-08-07 12:09:38.841 ERROR 1 --- [ntainer#2-0-C-1] o.s.k.support.LoggingProducerListener : Exception thrown when sending a message with key='a6558a22-470d-4708-b29
浏览 1提问于2020-08-07得票数 6
回答已采纳
我是第一次接触Kafka和Kstreams。我正在尝试连接两个流，并将输出推送到第三个流。我已经尝试了几天的各种实现，现在仍然坚持这个错误。错误，无法继续。有人能帮帮忙吗？
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.k
浏览 0提问于2020-05-02得票数 2
我有大约500个连接器，每次我创建一个新的连接器，我必须等待很长时间才能重新创建以前的连接器。它太慢了。 sink.properties如下 {
"name": "saas-saas_order-worder_fee_price-sink",
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
"errors.log.include.messages
浏览 12提问于2019-06-12得票数 0
回答已采纳
1回答
据我所知，动物园管理员存储有关kafka的元数据，主要涉及领导选举、分区细节等。
但是当我删除/tmp/kafka-logs目录并重新启动kafka时，它再次在/tmp/kafka-logs目录中获取数据。
是否有可能动物园管理员也持有数据，并将其传递给卡夫卡来同步？
我使用的是卡夫卡的最新版本，是1.0.1
浏览 1提问于2018-03-28得票数 1
回答已采纳
我在分布式模式下运行一个Kafka connect worker。(这是一个测试集群)，我想要重置默认的connect-* topics，所以在不停止我删除的工作线程的情况下，在工作线程重新启动后，我得到了这个错误。 ERROR [Worker clientId=connect-1, groupId=debezium-cluster1] Uncaught exception in herder work thread, exiting: (org.apache.kafka.connect.runtime.distributed.DistributedHerder:324)
org.apa
浏览 98提问于2021-07-29得票数 0
回答已采纳
有人对文件库连接器的输出格式有问题吗？
我想知道为什么我会收到这样的格式
{op=u, before={withTax=1, unit=pc(s), deleted=0, updated_at=null, price=420000.0, name=Test 123, created_at=null, id=54, useTaxIncludedFormula=0}, after={withTax=1, unit=pc(s), deleted=0, updated_at=null, price=333.0, name=Test 123, created_at=null, id=54, useTax
浏览 0提问于2019-06-20得票数 0
回答已采纳
我正在尝试测试Brooklin，以便在kafka集群之间镜像数据。我正在关注维基https://github.com/linkedin/brooklin/wiki/mirroring-kafka-clusters 与维基不同的是，我尝试在两个不同的集群之间建立镜像。我能够启动Brooklin进程和数据流，但我不能设法镜像消息。Brooklin在源kafka群集ATM上运行。我正在尝试镜像主题'test‘。 布鲁克林的server.properties是 ############################# Server Basics ######################
浏览 31提问于2020-10-14得票数 0
回答已采纳
1回答
我正在尝试使用与星火结构的流查询。我得到的错误是，用户在处理数据时找不到合适的偏移量。如果没有此触发器，查询将正常运行(如预期的那样)。
我所做的：
阅读卡夫卡主题的数据：
val inputStream: DataStreamReader = spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "host:port")
.option("subscribe", "input_topic&
浏览 6提问于2021-10-06得票数 2
2回答
我正在使用KAFKA从一个来源获取数据。我想知道，如果我在微服务架构中串行增加组件数量，是否会使其更具可伸缩性？或者只是增加相同组件类型的KAFKA消费者的数量就会增加吗？对于任何错误，我很抱歉，我是第一次发布一个问题。
浏览 23提问于2020-04-27得票数 0
回答已采纳
目前，我有以下设置：
StoreBuilder storeBuilder = Stores.keyValueStoreBuilder(
Stores.persistentKeyValueStore("kafka.topics.table"),
new SomeKeySerde(),
new SomeValueSerde());
streamsBuilder.addStateStore(storeBuilder);
final KStream<byte[], SomeClass> requestsStream = streamsBuilde
浏览 4提问于2019-12-07得票数 2
回答已采纳
我们正在尝试使用Debezium从许多Postgresql数据库中收集更改。这样做的目的是创建一个主题，其分区数量等于数据库的数量--每个数据库都有自己的分区，因为事件的顺序很重要。我们成功地使用主题路由将事件重新路由到单个主题，但为了能够按数据库对事件进行分区，我需要正确设置消息键。
问:我们是否可以将kafka消息键设置为与源数据库名称相等？
我的想法：
可能有一种方法可以在消息中找到每个连接器configuration?Database名称全局设置消息键的方法，但这是一个嵌套的属性payload.source.name。没有找到从嵌套属性中提取值的方法.
有什么想法吗？
提前谢谢你！
浏览 9提问于2022-09-07得票数 0
2回答
我正在寻找从Kafka读取消息的最佳方法(每天大约100 B)，阅读了我需要对数据进行操作并将其写入HDFS的消息。
如果我需要以最好的性能完成这个任务，那么从Kafka读取消息并将文件写入HDFS的最佳方法是什么？
哪种编程语言最适合这一点？
我是否需要考虑使用像斯派克这样的解决方案？
浏览 0提问于2018-05-14得票数 2
1回答
我对kafka非常陌生，我试图将数据写到一个主题中，并从同一个主题中读取数据(我们现在作为一个源团队来摄取数据)。因此，我们在对Kafk主题进行写操作，并从相同的主题中使用)。我在星火壳上编写了下面的代码，以便将数据写入一个Kafka主题。
pyspark --packages io.delta:delta-core_2.11:0.6.1,org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.0,io.strimzi:kafka-oauth-client:0.5.0
from pyspark.sql.functions import col
from
浏览 3提问于2021-09-08得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
