多连接器一主题分布式Kafka连接
是指在分布式系统中，使用多个连接器将数据从不同的数据源连接到一个主题中的Kafka集群。
Kafka是一个高吞吐量、低延迟的分布式消息队列系统，常用于大规模数据处理和实时流数据处理。它采用发布-订阅模式，将数据以消息的形式进行传输和存储。
多连接器一主题的设计模式可以实现将来自不同数据源的数据集中到一个主题中，方便进行统一的数据处理和分析。这种模式可以应用于各种场景，例如日志收集、数据同步、事件驱动等。
优势：
- 数据集中管理：通过将不同数据源的数据连接到一个主题中，可以方便地进行数据管理和处理，减少了数据分散的问题。
- 灵活性：可以根据需求灵活地添加或移除连接器，实现动态的数据源扩展和收缩。
- 高可靠性：Kafka具有高可靠性和容错性，可以保证数据的可靠传输和存储。
- 高吞吐量：Kafka的设计目标之一就是高吞吐量，可以处理大规模的数据流。
应用场景：
- 日志收集和分析：将来自不同服务器的日志数据连接到一个主题中，方便进行日志的集中管理和分析。
- 数据同步：将不同数据库中的数据连接到一个主题中，实现数据的实时同步和备份。
- 实时流数据处理：将来自不同传感器或设备的实时数据连接到一个主题中，进行实时的数据处理和分析。
- 事件驱动架构：将不同系统的事件数据连接到一个主题中，实现系统之间的解耦和事件驱动。
推荐的腾讯云相关产品： 腾讯云提供了一系列与Kafka相关的产品和服务，可以帮助用户快速搭建和管理分布式Kafka集群。
- 云原生消息队列 CKafka：腾讯云提供的高可靠、高吞吐量的分布式消息队列服务，基于Apache Kafka开发，支持多连接器一主题的设计模式。 产品链接：https://cloud.tencent.com/product/ckafka
- 数据接入服务 DTS：腾讯云提供的数据传输服务，可以实现不同数据源之间的数据同步和迁移，支持将数据连接到Kafka主题中。 产品链接：https://cloud.tencent.com/product/dts
- 流计算服务 SCF：腾讯云提供的无服务器计算服务，可以实现实时流数据处理和分析，支持与Kafka集成。 产品链接：https://cloud.tencent.com/product/scf
请注意，以上推荐的产品仅为示例，其他云计算品牌商也提供类似的产品和服务。
相关·内容
我是开始卡夫卡连接在独立模式如下
/usr/local/confluent/bin/connect-standalone /usr/local/confluent/etc/kafka/connect-standalone.properties /usr/local/confluent/etc/kafka-connect-elasticsearch/quickstart-elasticsearch.properties
之后，我使用rest创建了一个包含所有细节的连接器。像这样
curl -X POST -H "Content-Type: application/json" -
浏览 7提问于2020-02-01得票数 1
1回答
我的问题分成两部分。我读过。我想要完全删除一个连接器，带有偏移和所有，这样我可以重新创建它的同名以后。这个是可能的吗？据我所知，墓碑信息将无限期地扼杀这个连接器。
第二部分是--是否有办法让kafka-connect容器在关闭时自动删除他创建的所有连接器？谢谢
浏览 7提问于2022-11-17得票数 1
1回答
我在centos 7(汇合)上安装了Apache，我试图在分布式模式下运行filestream连接，但是我得到了以下错误：
[2017-08-10 05:26:27,355] INFO Added alias 'ValueToKey' to plugin 'org.apache.kafka.connect.transforms.ValueToKey' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:290)
Exception in thread "main" o
浏览 6提问于2017-08-10得票数 1
回答已采纳
1回答
我正在使用这个Github回购和文件夹路径，我发现：运行卡夫卡本地独立模式连接。我对Docker撰写文件做了一些更改，但主要是与身份验证有关的更改。
我现在遇到的问题是，当我运行Docker映像时，对每个分区(有10个分区，从0到9)，我多次得到这个错误：
[2021-12-07 19:03:04,485] INFO [bq-sink-connector|task-0] [Consumer clientId=connector- consumer-bq-sink-connector-0, groupId=connect-bq-sink-connector] Found no commit
浏览 3提问于2021-12-07得票数 0
我在Kafka connect中有一个主题，我想把数据写到两个不同的位置。为此，我有两个自定义连接器。
我不是在使用汇合平台，我只是在研究卡夫卡集群，我计划使用卡夫卡的connect-distributed.sh。
但是，我不知道如何部署这些不同的工作。我应该创建两个distributed.properties文件，并将它们部署到不同的端口上吗？
另外，我应该将每个属性和连接器jar文件放到kafka集群的每个节点上吗？
谢谢。
浏览 0提问于2019-07-28得票数 0
是否可以使用单独的Kafka实例与Elasticsearch连接器一起写入使用相同索引？的Elasticsearch集群。源数据可以是后端数据库或应用程序。一个示例用例是，一个集群可以用于实时搜索，另一个可以用于分析。
如果这是可能的，我如何配置接收器连接器？如果没有，我可以想出几个选择：
使用2个Kafka实例，每个实例指向不同的Elasticsearch集群。要么两者都写，要么写到其中一个，然后从其中复制到另一个。
使用一个Kafka实例并编写一个流处理器，它将写入两个集群。
还有其他的吗？
浏览 3提问于2020-11-17得票数 0
回答已采纳
1回答
我有使用独立的方法，而不是通过汇合的过程或作为整个平台的一部分。
我可以通过以下命令成功地从命令行启动连接器：
./kafka_2.11-2.1.0/bin/connect-standalone.sh connect.properties s3-sink.properties
可以看到来自的主题CDC偏移正在被消耗。不会抛出错误。但是，在AWS S3中，不为新数据创建文件夹结构，也不存储JSON数据。
问题
连接器是否应该在看到主题的第一个JSON数据包时动态创建文件夹结构？
除了配置awscli凭据之外，connect.properties和s3-sink.properties还有其
浏览 0提问于2019-04-24得票数 1
回答已采纳
使用一个由3组成的Kafka集群和一个相同的Zoo门将集群，我提出了一个分布式连接器节点。此节点仅通过一个任务成功运行。然后，我提出了第二个连接器，这似乎是在任务中的一些代码确实运行时运行的。然而，它似乎没有存活(虽然没有抛出错误，但由于缺少预期的活动而观察到不活动，而第一个连接器继续正常工作)。当我在每个连接器节点上调用http://localhost:8083/connectors/mqtt/tasks时，它会告诉我连接器只有一个任务。我希望这是两个任务，每个节点/工作人员一个任务。(目前的工作人员配置说是tasks.max = 1，但我也尝试将它设置为3。
当我试图打开第三个连接器时，我
浏览 4提问于2016-12-21得票数 11
回答已采纳
我在使用汇合JDBC连接器时遇到了非常奇怪的行为。我很确定这与汇流堆栈无关，而是与Kafka-connect框架本身有关。
因此，我将offset.storage.file.filename属性定义为默认/tmp/connect.offsets并运行我的接收器连接器。显然，我期望连接器在给定的文件中持久化偏移量(它在文件系统中不存在，但是它应该自动创建，对吗？)文件上说：
offset.storage.file.filename用于存储连接器偏移量的文件。通过在磁盘上存储偏移量，可以在单个节点上停止并启动独立进程，并恢复到以前停止的位置。
但卡夫卡的行为方式完全不同。
它检查给定文
浏览 3提问于2017-02-06得票数 13
1回答
我试图使用AWS DocumentDB作为存储从Kafka接收到的数据的接收器，并想知道MongoDB Kafka连接器是否与DocumentDB一起工作，因为它的文档提到它与MongoDB驱动程序兼容。
如果不是这个连接器，除了建立一个定制的kafka连接之外，还有什么替代的方法呢？
浏览 6提问于2022-05-31得票数 0
1回答
我正忙于学习kafka，特别是使用接收器连接器(jdbc)将数据从kafka主题发送到mysql的下游元素。
我已经在这样的Dockerfile文件中，使用jdbc连接器和mysql驱动程序构建了一个由融合式kafka连接基映像生成的映像：
FROM confluentinc/cp-kafka-connect-base
RUN confluent-hub install --no-prompt confluentinc/kafka-connect-jdbc:5.4.1
ENV MYSQL_DRIVER_VERSION 5.1.39
RUN curl -k -SL "https:/
浏览 3提问于2020-03-19得票数 0
回答已采纳
我想知道如何将数据从elasticsearch发送到kafka，然后再发送到influxdb？我已经尝试过使用来自elasticsearch的源连接器和从influxdb的接收器连接器的汇合平台，但问题是我不得不将数据从elasticsearch发送到kafka
而且，一旦我的电脑关机，我就不再有连接器的备份，我必须从头开始。
所以我才会问：
如何将数据从elasticsearch发送到kafka？使用汇流平台？
如果我想使用卡夫卡连接，我真的必须使用汇合平台吗？
浏览 3提问于2022-09-15得票数 0
回答已采纳
我使用MongoSourceConnector将kafka主题与mongo数据库集合连接起来。对于带有单个kafka主题的单个数据库，它工作得很好，但是有什么方法可以用单个kafka主题连接多个mongo数据库吗？
浏览 3提问于2021-11-25得票数 0
Kafka连接器可以使用主键和时间戳来确定需要处理哪些行。
我正在寻找一种方法重置连接器，以便它将处理从开始的时间。
浏览 3提问于2017-03-24得票数 7
回答已采纳
1回答
我正在遵循本指南中的步骤
我得到的错误信息是
BadRequestException:连接器配置{.}不包含连接器类型
我在运行命令时
sh kafka_2.12-2.3.0/bin/connect-standalone.sh connect-standalone.properties snowflake_kafka_config.json
我的配置文件是
connect-standalone.properties
bootstrap.servers=localhost:9092
value.converter=org.apache.kafka.connect.json.JsonConvert
浏览 0提问于2019-10-15得票数 0
回答已采纳
我想从各种S3存储桶(超过10,000个存储桶，每个文件大约20-50MB)加载数据到Apache Kafka。存储桶列表是动态的-在运行时添加和删除存储桶。理想情况下，每个存储桶配置都应该有自己的轮询间隔(扫描新文件的频率-至少60秒，但可能更长)和优先级(正在处理的并发文件的数量)。 请注意，由于每个存储桶所有者的组织中存在各种IT策略，因此无法设置从每个S3存储桶到SQS/ in /Lambda的通知。 Kafka Connect似乎是这类任务最常用的工具，它的可插拔架构将使未来添加新源代码变得更容易，因此它非常适合。将每个S3存储桶配置为其自己的连接器将允许我为每个任务设置不同的任务
浏览 49提问于2021-07-28得票数 2
我已经在windows的docker上安装了kafka，它的运行方式如下面的截图所示，我已经使用这个链接安装了“亚马逊S3源连接器”。
我的问题是：
I)如何执行命令来查看所有kakfa主题(不通过GUI)
ii)如何检查S3源连接器是否已安装属性(我指的是位置或CLI.)
iii)和link还指定创建和使用“quickstart-s3soure.properties”，我在哪里可以做到这一点？在我桌面上的docker上？
浏览 1提问于2021-04-24得票数 0
当我使用镜像器2时，我有一个问题。
Mirrormaker 2基于Kafka框架，可以将其核心看作是Kafka源和接收器连接器的组合。因此，在MM2体系结构中，有源和接收器连接器。但是，对于MM2中的连接器，是否有额外的Kafka集群？因为在kafka连接设计中，源和汇连接器需要Kafka集群来移动数据。
例如，MM2需要源和目标集群；我的问题是，需要第三个MM2来进行镜像操作，而不使用源集群和目标集群。
另一个问题是，是否可以在分布式模式下运行MM2连接器? --我没有看到关于这个问题的任何配置？
例如，在docker环境中；下面的配置是否足以在分布式模式下运行MM2？
mirrormake
浏览 5提问于2020-10-15得票数 0
我试图运行卡夫卡连接的分布式模式，从主题读取和写入HDFS。
该过程正在成功地启动，提供了所需的属性文件，但我无法将事件提交给HDFS。
请看下面的完整日志。
注意:代理详细信息已被屏蔽，日志中提到了所有其他细节。
[2018-06-26 11:49:48,474] INFO ConsumerConfig values:
auto.commit.interval.ms = 5000
auto.offset.reset = earliest
bootstrap.servers = [****:****, ****:****]
che
浏览 2提问于2018-06-26得票数 0
1回答
我正在设计一个必须与基于多个消息队列(MQ)的后端集成的系统。我为每个后端提供了一个微服务，用于处理MQ有效负载。我选择了Kafka作为消息传递的媒介，并考虑将Kafka- MQ -Connects用于MQ集成。
我可以想到两种集成方法。
Kafka -(源/库)连接每个后端+Kafka主题(到/从)每个后端。
专业人士。-可以扩展到新的后端，而无需接触现有的连接器。
缺点。-太多的连接器和主题需要维护。
单个Kafka -(来源/水槽)+单个Kafka主题(对/从)所有后端。此外，Sink连接到MQs的动态路由，而微服务将内置消息过滤器，只过滤相关的消息。
浏览 0提问于2019-07-19得票数 2
我将在本地“启动”Docker容器(独立于Kafka节点容器)中以分布式模式启动Kafka连接器。连接器按预期工作，但当我关闭启动容器时，连接器将停止工作。我希望它能继续工作，因为我相信它是在一个不同容器的Kafka节点上的一个工人上注册和运行的。我的详细设置如下：
目前，我正在通过码头集装箱本地运行所有的东西。我有：
动物园管理员节点(3.4.9)
一个Kafka节点(Apache，0.10.1.0)
发射节点。
启动节点下载适当的Kafka版本并解压缩其内容。然后构建连接器源，设置类路径以包含必要的JAR，然后以如下方式执行连接器：
connect-distribute
浏览 1提问于2016-12-06得票数 4
回答已采纳
我们正试图在给定节点上启动多个独立的kafka hdfs连接器。
对于每个连接器，我们分别将rest.port和offset.storage.file.filename设置为不同的端口和路径。
同样，kafka broker JMX端口是@ 9999。
启动kafka独立连接器时，会出现错误。
错误:代理引发的异常: java.rmi.server.ExportException:已在使用的端口: 9999；嵌套异常为: java.net.BindException:已在使用的地址(绑定失败)
尽管rest.port设置为9100
kafka版本: 2.12-0.10.2.1
kafk
浏览 7提问于2017-06-14得票数 2
我有卡夫卡连接(版本2.1.1-cp1)与卡夫卡(2.0.1-cp4)和运行大约70个连接器相当繁重的工作量。有时(每2-3周)，突然间，一些节点开始获得以下日志并停止工作：
INFO [Worker clientId=connect-1, groupId=dwh-prod] (Re-)joining group (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
INFO [Worker clientId=connect-1, groupId=dwh-prod] Successfully joined group
浏览 3提问于2020-08-25得票数 0
我想知道在confluent center中使用data gen连接器的过程，这样我就可以为测试主题生成随机数据。
我尝试使用某些设置连接datagen连接器，但总是失败。
更新:我尝试连接在topic中创建的我自己的模式，但它不能使用它，datagen连接器无法启动。
下面是datagen连接的配置：
{
"value.converter.schema.registry.url": "http://localhost:8081",
"value.converter.schemas.enable": "false",
浏览 0提问于2020-08-05得票数 0
kafka connect是否在kafka connect流程中启动新的连接器及其任务？否则将派生一个新的JVM进程。
如果它在kafka连接过程中启动插件，那么我需要通过KAFKA_CONNECT_JVM_HEAP_OPT设置kafka connect JVM堆的大小(使用confluent docker镜像)。然后，问题是，如果我启动了许多任务或连接器，它们将共享JVM堆，因此很难决定kafka connect的堆大小。
如果对于每个连接器，kafka connect在新的JVM进程中启动它们，我如何为它们设置堆大小？
浏览 4提问于2018-05-31得票数 4
我们已经使用jdbc独立连接器成功地使用了MySQL - kafka数据摄取，但现在在分布式模式(作为kafka connect服务)中使用同样的连接器时遇到了问题。
用于工作正常的独立连接器的命令-
/usr/bin/connect-standalone /etc/kafka/connect-standalone.properties /etc/kafka-connect-jdbc/source-quickstart-mysql.properties
现在我们已经停止了这个服务，并在分布式模式下启动了kafka连接服务，如下所示-
systemctl status confluent-kaf
浏览 1提问于2018-11-15得票数 2
我正在尝试将一个postgreql (它是源)事件日志汇到一个已经创建和配置的kafka汇流云上。
FYI，我使用quay.io/debezium/connect:1.9停靠映像作为我的kafka连接服务器，使用aws数据库作为源，使用kafka汇合云主题作为接收器。
我知道属性中有CONNECT_TOPIC_CREATION_ENABLE=false = topic.creation.enable: false这样的参数可以阻止自动创建kafka主题(因为它在我的合流云代理中是不允许的)，但是我在哪里指定主题目的地名称呢？
浏览 4提问于2022-05-25得票数 0
回答已采纳
根据阅读kafka连接文档：
config.storage.topic=connect-configs
bin/kafka-主题-创建-动物园管理员本地主机:2181-主题连接-信任-复制-因子3-分区1 -config cleanup.policy=compact
offset.storage.topic=connect-offsets
bin/kafka-主题-创建--动物园管理员本地主机:2181-主题连接-偏移-复制-因子3-分区50 -config cleanup.policy=compact
status.storage.topic=connect-status
bin/kafk
浏览 3提问于2022-05-17得票数 0
回答已采纳
我想在使用Kafka Connect Source JDBC连接器导入表格时设置mesage键。
当有多个表被定义为从JDBC连接器读取时，Kafka Connect/Source中的单个消息转换(SMT)如何针对正确的字段？SMT需要一个列名，当有多个表时，列名可能会有所不同。
我没有看到一种根据表名或类似名称过滤SMT定义的方法。下面的代码示例运行良好，因为它只是一个表。
但是，如果您有不同的表，例如User、Order、Product，该怎么办？
"table.whitelist" : "User"
"transforms":"cr
浏览 12提问于2019-07-13得票数 2
回答已采纳
1回答
我们一直试图在AWS机器上建立一个生产级Kafka集群，但到目前为止，我们一直没有成功。
Kafka版本: 2.1.0
机器：
5 r5.xlarge machines for 5 Kafka brokers.
3 t2.medium zookeeper nodes
1 t2.medium node for schema-registry and related tools. (a Single instance of each)
1 m5.xlarge machine for Debezium.
默认代理配置：
num.partitions=15
min.insync.replicas=1
浏览 0提问于2018-12-13得票数 5
我有一个运行的debezium设置，用于做MySQL的疾病预防控制中心。现在，我想为另一个MySQL服务器再创建一个MySQL连接器。但是我不想要现有数据的快照，我希望从特定的文件和位置启动debezium新连接器。
我读了一些来自堆栈溢出的问题，他们告诉我手动将记录插入到connect-offsets主题中。但是，如果我这样做，在测试服务器上的现有安装程序上会发生什么情况，我尝试设置上面的解决方案，但它没有工作。
kafka-console-producer --broker-list localhost:9092 --topic connect-offsets
>{"file
浏览 3提问于2019-12-30得票数 1
目前，我设置了两个独立的连接器，运行JDBC Sink Connector来摄取生产者生成的主题，并将其读入数据库。有时，我在日志中看到错误，这会导致生成的消息无法存储到数据库中。我经常看到的错误是
Caused by: org.apache.kafka.common.errors.SerializationException: Error retrieving Avro schema for id:11
Caused by: io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: Subjec
浏览 22提问于2021-10-29得票数 0
我想设置从kafka主题到cassandra的kafka连接
问题很简单:假设我在kafka中有一个演示主题，其中包含json数据，比如
{"id":"1", "name":"Alex", "clicks":2}
我想自动将它推入到cassanra表中，列有id，name，click。
我正在研究，但我能找到的唯一示例是从cassandra读取，并通过中间的kafka写入另一个cassandra表。
我的问题是，我怎么才能让它读卡夫卡而不是卡桑德拉呢？我正在寻找一些连接器开源的例子来做这件事。
浏览 1提问于2016-07-21得票数 2
我想在Apache Kafka中创建实时数据管道。我有一个数据库，它位于远程位置，该数据库不断更新。我应该使用哪个Kafka连接API从数据库中提取数据并实时摄取到Kafka broker中？稍后，我将使用kafka stream和KSQL运行即席查询来执行指标。
任何帮助都将不胜感激！
浏览 2提问于2018-08-28得票数 1
1回答
我使用的是由卡夫卡连接而成的Debezium。Kafka Connect公开了一些需要创建的主题：
OFFSET_STORAGE_TOPIC在运行Kafka服务时需要这个环境变量。将此设置为Kafka主题的名称，在该主题中，组存储连接器中的Kafka服务会偏移。主题应该有许多分区，高度复制(例如，3x或更多)，并且应该配置为压缩。
STATUS_STORAGE_TOPIC在运行Kafka服务时应该提供这个环境变量。将此设置为Kafka主题的名称，在该主题中，组存储连接器状态中的Kafka服务。主题可以有多个分区，应该被高度复制(例如，3x或更多)，并且应该配置为压缩。
是否有人对这些主题有任何
浏览 4提问于2022-05-03得票数 0
回答已采纳
2回答
我想在我们的微服务中实现发件箱模式，使用Mongo连接器，在发件箱( MongoDB集合)中，我使用以下字段存储主题数据：kafka_topic、kafka_key、kafka_value、kafka_header_keys、kafka_header_values。
但是，我应该如何配置mongo连接器来动态地从kafka_topic字段中选择主题，以及如何从其他发件箱字段中选择其值和头呢？我无法从它的配置引用中找到进行此操作的设置。
我还在考虑从mongo连接器存储库获得一个叉子，并使用我的自定义实现来扩展它。
如果你能帮我，我会非常感激的。
浏览 3提问于2020-12-25得票数 2
我有一个在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的HDFS连接器配置为将几个单独的主题写入一个文件中？
这些主题将包含具有相同avro模式的消息，我希望KafkaConnect充当这些Kafka主题和HDFS之间的中介。在最坏的情况下，主题内容可以在编写到HDFS之后结合起来，但是我觉得使用HDFS连接器应该是一种更干净、更快捷的方法。
浏览 3提问于2016-08-31得票数 1
回答已采纳
1回答
我需要在不同的数据库上使用两个完全相同的表(位置也不同)才能在同一主题上发布。Kafka JDBC连接器没有解释它是如何管理高水印的，所以我们想检查一下这个场景中的最佳实践是什么? 1.我们是否可以将两个单独的JDBC连接器发布到单独的主题2。我们是否可以保持两个单独的JDBC连接器发布到同一个主题。
如果我们选择选项2，那么Kafka连接器是如何在消息同时到达表的情况下管理的呢？如何管理不同的数据库时区？
浏览 0提问于2020-04-06得票数 0
回答已采纳
1回答
我正在做一个项目，使用kafka连接从多个数据库源拉出数据。然后，我希望能够将数据转换为指定的json格式，然后最终将最终的json推送到S3存储桶中，最好使用kafka connect来降低开销。
以下是当前数据进入kafka的示例( avro格式)：
{"tableName":"TABLE1","SchemaName{"string":"dbo"},"tableID":1639117030,"columnName":{"string":"DATASET"}
浏览 7提问于2017-07-06得票数 2
回答已采纳
我希望使用Source连接器将SQL中的数据导入kafka主题，因此，为了配置源代码属性文件，我将从SQL表中获取整个数据到Kafka主题中，但我不希望从SQL表中获得全部数据，我希望从表中获得特定的列，表可能是多个，主要是我不知道如何配置每个查询的多个主题。
浏览 1提问于2019-10-03得票数 0
回答已采纳
从Kafka connect中删除源连接器后，为该连接器保存的偏移量仍然存在于Kafka中。当具有相同名称的相同连接器再次启动时，它将从上次保存的偏移量开始处理数据。更改连接器的名称不是一个解决方案，因为它使得连接器提交很难自动化。 是否有解决此问题的方法(重命名源连接器不是一个选项)
浏览 77提问于2021-06-17得票数 2
回答已采纳
对于CDC来说，在服务器上运行Kafka集群的最佳方法是什么？
如何将mysql db连接到运行在本地的kafka集群，并创建到另一个mysql db的接收器连接。
我认为REST是最好的方法，但不确定如何做。
我需要在生产者端运行配置吗？需要知道如何实现它
浏览 5提问于2022-01-19得票数 0
回答已采纳
我使用的是kafka_2.11.-0.9.0.1。
首先，我用以下命令启动了zookeeper & kafka (如所解释的)：
bin/zookeeper-server-start.sh config/zookeeper.properties
然后：
bin/kafka-server-start.sh config/server.properties
现在，我正在尝试从REST API (如所解释的)中获取一些信息：
[itay@localhost general]$ wget http://127.0.0.1:8083/connectors
--2016-04-27 10:44:2
浏览 0提问于2016-04-27得票数 1
我在EC2上设置了一个汇合(4.0)连接器，它从kafka读取并写入S3。
独立尝试进行得很好：
bin/connect独立的etc/standalone/example-connect-worker.properties etc/standalone/example-connect-s3-sink.properties
但是，分布式版本始终失败。
[2018-01-30 21:26:05,860] ERROR Unexpected exception in Thread[KafkaBasedLog Work Thread - connect-configs,5,main] (org.ap
浏览 1提问于2018-01-30得票数 1
回答已采纳
正如标题所述，我使用debezium Postgres源连接器，并且我希望MongoDB接收器连接器根据不同的集合和数据库(也不同的dbs来分离无关数据)中的名称对kafka主题进行分组。在询问时，我在上看到了连接器属性。不幸的是，这只为每个kafka主题创建了一个集合，该集合成功地与指定的regex匹配，我计划使用相同的mongodb服务器来存储从多个debezium源连接器中捕获的许多dbs。你能帮帮我吗?
注意:我读过这个Mongo接收器设置FieldPathNamespaceMapper，但我不确定它是否符合我的需要，也不确定如何正确配置它。
浏览 14提问于2022-09-24得票数 0
我在kafka中创建了主题(test1，test2，test3)，我想在创建时将它们下沉到弹性。我尝试过topics.regex，但它只为已经存在的主题创建索引。当一个新的主题被动态创建时，我如何将它嵌入到一个索引中？
下面是我在kafka-sink中使用的连接器配置：
{
"name": "elastic-sink-test-regex",
"config": {
"connector.class": "io.confluent.connect.elasticsearch.Elastics
浏览 0提问于2018-10-02得票数 0
我试图使用FileStreamSink连接器备份卡夫卡数据。我知道有更好的选择，但我的公司已经有了文件备份基础设施(基于NetApp)，所以我想将Kafka数据转储到二进制文件并备份该文件。
因为在Kafka中存储的数据是加密的，所以我们没有一个模式可以使用或转换。我试着使用这个设置，但似乎不太好: key.converter=org.apache.kafka.connect.converters.ByteArrayConverter value.converter=org.apache.kafka.connect.converters.ByteArrayConverter
你对这个案子有什么
浏览 0提问于2019-10-29得票数 0
回答已采纳
我正在尝试使用debezium kafka连接以下设置的cdc：
ZookeeperKafka PostgreSQLKafka连接Debezium
我还按照这个顺序配置和启动它们。
启动Kafka Connect Debezium之后，我做了一个帖子来配置它：
{
"name": "postgres-boxes-connector",
"config": {
"connector.class": "io.debezium.connector.postgresql.PostgresConnector"
浏览 3提问于2019-11-27得票数 0
回答已采纳
Kafka接收器连接器提供3 insert.mode ..but，我需要更新或插入功能在一起。任何人都能帮助我们实现这一目标。
浏览 4提问于2021-12-20得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
