来自Kafka connect的MongoSource创建奇怪的_data密钥
Kafka Connect是一个用于连接Kafka和外部系统的工具，它允许将数据从外部系统导入到Kafka或将数据从Kafka导出到外部系统。MongoSource是Kafka Connect的一个插件，用于从MongoDB数据库中读取数据并将其发送到Kafka主题。
关于"奇怪的_data密钥"，这可能是指在使用MongoSource插件时出现的一个问题。在Kafka Connect中，配置文件中的密钥（key）通常用于指定数据的字段名，而"_data"可能是一个特定的字段名。然而，具体情况可能因使用的版本和配置而有所不同。
为了解决这个问题，可以尝试以下步骤：
- 检查配置文件：确保在配置文件中正确地指定了MongoSource插件的相关配置，包括MongoDB的连接信息、数据库和集合名称等。
- 检查数据模式：MongoSource插件通常会根据MongoDB中的数据模式自动创建Kafka消息的键值对。如果"_data"出现在数据模式中，可能是因为MongoDB中的某个字段名为"_data"，可以尝试修改MongoDB中的字段名，或者在配置文件中指定其他字段作为键值对的键。
- 查看日志信息：查看Kafka Connect的日志信息，以了解更多关于"_data"密钥的错误或警告信息。根据日志信息，可以进一步定位问题所在并采取相应的解决措施。
总之，对于Kafka Connect的MongoSource插件创建奇怪的"_data"密钥，需要仔细检查配置文件、数据模式和日志信息，以找出问题的根源并进行相应的调整和修复。
腾讯云相关产品和产品介绍链接地址：
- 腾讯云消息队列 CKafka：https://cloud.tencent.com/product/ckafka
- 腾讯云数据库 MongoDB：https://cloud.tencent.com/product/mongodb
相关·内容
我使用twitter使用python和kafka提取数据，我想在我的主题中创建两个分区，所以这是我的生产者文件中的代码：
for tweet in hashtag_tweets:
data = {}
data["content"] = tweet._json["full_text"]
data["retweets"] = tweet._json["retweet_count"]
data["favorites"] = tweet._json["favorite_count&#
浏览 8提问于2022-06-14得票数 0
1回答
Starting the connector:
curl -X POST -H "Content-Type: application/json" --data @snowflake-connector.json http://XX.XX.XX.XXX:8083/connectors
post executing the above command getting error:
{"error_code":500,"mess
浏览 13提问于2022-05-26得票数 0
我有一个卡夫卡Avro主题生成使用KafkaAvroSerializer。
我的独立属性如下。
我使用Confluent 4.0.0来运行Kafka connect。
key.converter=io.confluent.connect.avro.AvroConverter
value.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=<schema_registry_hostname>:8081
value.converter.schema.registry.ur
浏览 0提问于2018-06-28得票数 0
我使用卡夫卡连接创建一个MQTT卡夫卡连接，我把所有卡夫卡MQTT连接器特定的jar从汇流网站下载到"/data“文件夹。并相应地更新"connect-standalone.properties“文件以反映插件路径，即
plugin.path=/opt/kafka_2.11-2.1.1/libs,/data
当我运行卡夫卡连接
./connect-standalone.sh ../config/connect-standalone.properties ../config/connect-mqtt-source.properties
我得到以下错误：
2019-07-18 1
浏览 0提问于2019-07-18得票数 0
我试图从一个卡夫卡主题流到谷歌BigQuery。我的connect-standalone.properties文件如下：
bootstrap.servers=rhes75:9092,rhes75:9093,rhes75:9094,rhes564:9092,rhes564:9093,rhes564:9094,rhes76:9092,rhes76:9093,rhes76:9094
key.converter=org.apache.kafka.connect.storage.StringConverter
##value.converter=org.apache.kafka.connect.stor
浏览 1提问于2021-03-13得票数 0
回答已采纳
我们正在尝试使用Kafka连接到Snowflake数据库。根据下面的支持文档，我们已经创建了配置文件。我们使用JSON格式作为使用kafka的snowflake的有效负载输入。
支持文档：
配置文件：
{
"name":"FactoryData",
"Config":{
"connector.class":"com.snowflake.kafka.connector.SnowflakeSinkConnector",
"tasks.max":"8",
&#
浏览 1提问于2020-06-22得票数 0
我有一个用postgreSQL编写的接收器连接器，配置如下。
{
"name": "connector",
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
"connection.url": "jdbc:postgresql://localhost:5432/postgresuser=user&password=123",
浏览 2提问于2021-08-10得票数 0
我试图通过MySql将我的source-connector数据库更改外包给卡夫卡主题，这是可行的。现在我想把这些数据发送到弹性搜索实例。
为了做到这一点，我跟踪了这个和这个。
对于mysql到kafka的CDC，我可以看到我在mysql中所做的更改，并阅读它创建源-连接器，但是当我创建另一个连接器elasticsearch-sink连接器时，source-connectr task.state显示了Failed！因此，数据库更改不会进入ES，尽管在es-config.properties文件中作为安装程序在那里创建索引。
我已经将jar/s放在kafka中，为源-连接器的库工作(以避免关于类路
浏览 1提问于2020-02-19得票数 0
回答已采纳
我正在尝试使用debezium/connect:1.3 docker图像，并在尝试使用此tutorial创建弹性水槽时遇到以下错误。 正如this撰写文件所建议的那样，使用似乎不可用的debezium/connect-jdbc-es。 {
"error_code": 500,
"message": "Failed to find any class that implements Connector and which name matches io.confluent.connect.elasticsearch.Elasticsear
浏览 89提问于2020-12-29得票数 1
回答已采纳
我正在为kafka ()和JsonEventParser使用hbase接收器连接器，这里我遇到了一些问题。连接器返回错误：
org.apache.kafka.connect.errors.DataException: JsonConverter with schemas.enable requires "schema" and "payload" fields and may not contain additional fields. If you are trying to deserialize plain JSON data, set schemas.e
浏览 0提问于2018-12-07得票数 0
回答已采纳
1回答
亲爱的们： 我不知道我的配置出了什么问题。当我想启动Kafka服务时，我得到了一个错误： [2020-10-26 09:27:33,359] DEBUG Created SSL context with keystore SecurityStore(path=/etc/pki/CA/certs/hostname.jks, modificationTime=Mon Oct 26 08:59:45 CET 2020), truststore SecurityStore(path=/etc/pki/ca-trust/extracted/java/cacerts, modificationTime=
浏览 19提问于2020-10-26得票数 0
回答已采纳
我想使用filepulse连接器将xml文件加载到kafka。
下面是我的环境
Ubuntudownloaded Win10 WSL，安装了汇合平台5.5.1 (见“”)，unpackeddownloaded压缩文件版本1.5.2来自github ()，unzippedmodified的"connect-standalone.properties“位于汇流路径下(etc/kafka/connect-standalone.properties)，以包含path "/home/min/streamthoughts-kafka-connect-file-pulse-1.5.2/lib
浏览 5提问于2020-09-08得票数 0
我想使用kafka FileStreamSourceConnector将本地的avro文件写成一个主题。
我的连接器配置如下所示：
curl -i -X PUT -H "Content-Type:application/json" http://localhost:8083/connectors/file_source_connector/config \
-d '{
"connector.class": "org.apache.kafka.connect.file.FileStreamSourc
浏览 5提问于2021-03-26得票数 2
我正在设计一个MySQL -> Debezium -> Kafka -> Flink -> Kafka -> Kafka Connect JDBC -> MySQL。下面是我在Flink上写的示例消息(我也尝试过使用Kafka控制台生产者) {
"schema": {
"type": "struct",
"fields": [
{
"type": "int64",
"optional"
浏览 70提问于2019-02-27得票数 0
回答已采纳
1回答
我问这个问题是因为在最初的案例中没有答案：。
我也有类似的情况。
弹性搜索表可以为单个字段创建记录，但在通过kafkaconnect发送请求时不能创建多个字段。
获取异常“键”作为文档id，在弹性搜索中不能为空“”。
我的连接器配置：
{
"name": "test-connector33",
"config": {
"connector.class":"io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
"tasks.ma
浏览 1提问于2018-08-21得票数 0
我们让Kafka connect在Kube pod上运行。我们在worker日志中看到以下错误。我们试着重新启动吊舱。
[2020-04-02 14:40:13,237] WARN Aborting multi-part upload with id '0vbfuRZCRIkc431LJN.hyaPuo3ZQzuAsfTrMSdBE_Q9.sZP8g-' (io.confluent.connect.s3.storage.S3OutputStream)
[2020-04-02 14:40:13,274] ERROR Multipart upload failed to compl
浏览 0提问于2020-04-02得票数 0
1回答
我是apache环境的新手，它有一些让我很好奇的东西。我的kafka是通过ambari安装的。所以一切都可以通过ambari的GUI启动。当kafka通过ambari安装时，它包含kafka-connect。但是如果没有包含我需要的连接器怎么办？下面是一些包含的连接器。 /etc/kafka-connect-activemq
/etc/kafka-connect-elasticsearch
/etc/kafka-connect-ibmmq
/etc/kafka-connect-jdbc
/etc/kafka-connect-jms
/etc/kafka-connect-replicator
浏览 13提问于2020-06-12得票数 0
回答已采纳
我正在使用这个融合的protobuf示例：
根据这个Protobuf文件生成“Term”类型的消息：
syntax = "proto3";
package confluent.kafka.examples.protobuf;
message Section {
string Id = 1;
string SectionName = 2;
}
message Term {
string TermId = 1;
string TermName = 2;
repeated Section Section = 3;
}
这些消息需要找到它们到达Elasticsear
浏览 52提问于2021-07-09得票数 0
我使用mongo源代码来监听mongo change stream，并将所有事件放入kafka中，但我绞尽脑汁想要找到一种从事件中提取“真正”密钥的方法。我尝试了转换，但它不起作用，给了我错误：
Caused by: org.apache.kafka.connect.errors.DataException: Only Struct objects supported for [copying fields from value to key], found: java.lang.String
在Mongo源码中，我找到了这个
这基本上意味着它甚至没有一些键处理，相反，它查找"_id“
浏览 5提问于2019-12-24得票数 2
1回答
我正在尝试使用Kafka文件脉冲连接器()将数据从文件中读取到一个Kafka主题中。
我使用以下方法启动连接器：
../bin/connect-standalone filepulse.properties connect-standalone.properties
Filepulse.properties的内容：
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter
offset.storag
浏览 3提问于2021-01-01得票数 2
回答已采纳
我试图将Kafka主题键映射为文档id，同时使用Kafka接收器连接器对弹性搜索进行索引。但我得到一个异常，键为空。
正如在下面的示例中提到的，我希望将"AKA-25“键作为文档id，但是我正在做的转换失败了。如果我从值中提取任何其他字段，就像我尝试使用"empId“一样，它似乎工作得很好。但我的要求是映射钥匙。
有人能给我建议吗？
这里是连接器配置：
{
"connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
"typ
浏览 1提问于2021-03-10得票数 2
1回答
我有两个服务器，一个是UI服务器，另一个是kafka服务器。
我的UI服务器中的javascript文件正在从csv文件中获取数据，我正在逐行读取该文件并转换为JSON。我需要将这些JSON格式的逐行读取数据发送到我的Kafka生产者服务器。为了进一步的工作。两台服务器都有自己的专用IP地址。例如kafka服务器有192.168.2.12:9098
reportJSON是我在UI服务器js文件中获取csv数据的变量。
当我试图运行ui服务器的js文件时，它显示了错误：
2018-05-09T15:18:56.147Z -错误: uncaughtException: io.connect不是一
浏览 0提问于2018-05-09得票数 1
回答已采纳
在我的一个.net项目中，我使用带RSA算法的Bouncy城堡( BC )，我使用BC创建了根证书( root.crt )，我还使用BC创建了由root.crt签名的证书(server.crt)。
使用Keytool，我将server.crt添加到密钥存储库(server.keystore.jks)中，将root.crt添加到信任存储库(server.truststore.jks)。请查找以下命令，用于将这些证书推送到相应的商店。
将Server.crt导入server.keystore.jks:
密钥工具-keystore server.keystore.jks -validity 365
浏览 6提问于2021-06-17得票数 0
环境(Docker)： # 5.5.1
image: confluentinc/cp-zookeeper:latest
# 2.13-2.6.0
image: wurstmeister/kafka:latest
# 5.5.1
image: confluentinc/cp-schema-registry:latest
# 5.5.1
image: confluentinc/cp-kafka-connect:latest
# 0.11.0
image: confluentinc/ksqldb-server:latest Kafka主题内容来自kafka connect (使用debez
浏览 84提问于2020-08-29得票数 2
我想给Kafka服务器发个消息。代理列表不包括localhost。但是当Producer调用send方法时，它有一个异常:Producer连接到localhost:9092不成功
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import java.util.Properties;
// create a producer
Properties props = new Properties();
props
浏览 0提问于2013-07-23得票数 3
正在尝试根据以下链接的说明向我们的kafka集群添加更多的kafka连接器。但由于errImagepull错误而失败。请找到详细信息并帮助我解决此问题。 参考链接： https://docs.confluent.io/home/connect/extending.html#create-a-docker-image-containing-c-hub-connectors 已创建自定义Docker镜像： FROM confluentinc/cp-server-connect-operator:6.0.0.0
USER root
RUN confluent-hub install --no-
浏览 24提问于2020-11-28得票数 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个字段) AVRO格式的KSQL表作为JDBC连接器接收器(mysql)的源
我收到以下错误(在INFO行之后)：
[2018-12-11 18:58:50,678] INFO Setting metadata for table "DSB_ERROR_TABLE_WINDOWED" to Table{name='"DSB_ERROR_TABLE_WINDOWED"', columns=[Column{'MOD_CLASS', isPrimaryKey=false, al
浏览 2提问于2018-12-12得票数 0
1回答
我们有一个源连接器，从rdbms读取并放入kafka。它使用模式注册表和avro模式。我在kafka连接日志和模式注册表日志中分别发现了以下异常。
1.
Committing offsets (org.apache.kafka.connect.runtime.WorkerSourceTask:426)
WorkerSourceTask{id=A-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask:443)
Task threw an
浏览 19提问于2021-03-15得票数 0
Kafka连接器为消息提供了一个Null指针异常，其中模式在“parentId”中有一个可选字段。我错过了什么吗？我正在使用开箱即用的JSONConverter和JDBC Sink连接器。
关于卡夫卡主题的信息是
{
"schema":{
"type":"struct",
"fields":[
{
"field":"id",
"type":"string"
},
{
"fie
浏览 0提问于2019-01-11得票数 2
回答已采纳
1回答
在使用正式文档配置kafka连接之后.
我得到一个错误，驱动程序不存在内卡夫卡连接！
我必须尝试将.jar复制到上述目录，但什么也没有发生。
有什么解决办法吗？
码头组成
---
version: '2'
services:
zookeeper:
image: confluentinc/cp-zookeeper:latest
hostname: zookeeper
container_name: zookeeper
ports:
- 2181:2181
environment:
ZOOKEEPER_CLIENT
浏览 0提问于2021-05-19得票数 0
回答已采纳
我使用docker运行kafka，在卷上设置存储数据。我设置了一些源连接器，主题是用cleanup.policy delete自动创建的。通过使用kafka经理，我将策略更改为compact。
问题：
在停止/启动停靠-组合主题出现后，但是cleanup.policy被恢复回delete。
问题：
重启后如何持久化主题配置？
附加信息
我命令重新启动卡夫卡码头工人：
rm /kafka/data/1/meta.properties; docker-compose down && docker-compose up -d --no-recreate
Docker-compose.
浏览 0提问于2020-10-21得票数 1
回答已采纳
我正在尝试向具有2个分区的单个主题生成一些消息。所有消息都将只发送到2号分区。我希望生产者流会跨所有分区分发消息。 const kafka = require('kafka-node')
const { Transform } = require('stream');
const _ = require('lodash');
const client = new kafka.KafkaClient({ kafkaHost: 'localhost:9092' })
, streamproducer = new kafka
浏览 16提问于2020-01-07得票数 0
我有一个mysql表如下所示：
我使用kafka连接器将此表添加到kafka主题：
ksql> CREATE SOURCE CONNECTOR SOURCE_MYSQL_01 WITH (
'connector.class' = 'io.debezium.connector.mysql.MySqlConnector',
'database.hostname' = 'mysql',
'database.port' = '3306',
'database.
浏览 2提问于2020-07-14得票数 0
回答已采纳
2回答
我目前正在配置kafka连接(使用)，我使用环境变量成功地将它连接到Kafka：
docker run -it --rm --name AAAAAA-kafka-connect -p 8083:8083 \
-v aaaaa.jks:aaaaa.jks \
-v bbbbbb.jks:bbbbbb.jks \
-e LOG_LEVEL=INFO \
-e HOST_NAME="AAAAAA-kafka-connect" \
-e HEAP_OPTS="-Xms256m -Xmx2g" \
-e BOOTSTRAP_
浏览 14提问于2022-05-24得票数 0
回答已采纳
1回答
我构建了一个kafka connect docker镜像，并使用rancher启动了kafka connect cluster.The集群，需要连接启用了kerberos的hdfs。所以我需要把keytab文件放到集群中。我不想将我的密钥表文件构建到我构建的映像中，因为它很重要而且安全。因此，我想使用牧场主存储服务，但我不能将我的密钥表放入存储中？我的问题是:如何在牧场主或码头管理我的安全文件？
浏览 0提问于2017-05-27得票数 0
1回答
Kafka流我正在尝试流式传输，但我有一些问题，它不工作。首先，我有三个连接器，但我不能使用我自己的键。我需要钥匙才能加入他们对吧？如何使用2个或2个以上的密钥加入？我尝试复制如下内容: select * from (select a. * from users a内部连接deps b on a.dep = b.dep和a.group = b.group )a内部连接user_afy on a.id = b.id 我希望将数据保存在一个主题中的内部联接中，并将其用于外部联接。这是我的一个例子。 连接器属性： ....
connector.class=io.confluent.connect.
浏览 7提问于2020-01-15得票数 3
我的目标是使用kafka以json格式读取字符串，对字符串执行筛选，然后将消息(仍然是json字符串格式)清空。
出于测试目的，我的输入字符串消息如下：
{"a":1,"b":2}
我的执行守则是：
def main(args: Array[String]): Unit = {
// parse input arguments
val params = ParameterTool.fromArgs(args)
if (params.getNumberOfParameters < 4) {
println("Missing parameter
浏览 2提问于2017-08-08得票数 0
回答已采纳
我使用的是带有查询模式的jdbc源连接器，似乎没有指定表名的情况下，在模式注册表中为记录键和记录值注册的模式具有空的模式名称，并被分配了默认名称"ConnectDefault“，如Confluent的AvroData类中所定义的那样
当使用生成的avro源代码和SpecificAvroSerde运行Kafka Streams应用程序时，我收到以下错误：
Exception in thread "streams-app-6e39ebfd-db14-49bc-834f-afaf108a6d25-StreamThread-1" org.apache.kafka.streams
浏览 4提问于2017-11-03得票数 2
我的工作是汇合管理培训和运行实验室在码头桌面。PFA的码头-组成yaml文件。汇合控制中心在brower不开放。我正在使用打开。它以前是开着的，但现在不开了。我在我的电脑中所做的唯一改变就是安装McAfee Live。我甚至试图关闭防火墙，但这也没有帮助。
如果你有类似的经验，以及你是如何克服这个问题的，谁能分享一下吗？
docker-compose.yaml文件。
version: "3.5"
services:
zk-1:
image: confluentinc/cp-zookeeper:5.3.1
hostnam
浏览 5提问于2020-11-01得票数 4
2回答
我有一个pyspark应用程序，它使用来自Kafka主题的消息，这些消息由org.apache.kafka.connect.json.JsonConverter序列化。我使用confluent Kafka JDBC连接器来做这件事 问题是，当我使用消息时，ID列以某种编码文本形式出现，例如"ARM=“，而它应该是数字类型。 下面是我现在拥有的代码 spark = SparkSession.builder.appName("my app").getOrCreate()
sc = spark.sparkContext
sc.setLogLevel('WARN
浏览 20提问于2019-03-09得票数 1
回答已采纳
我有一个码头组成的群与动物园管理员服务启动良好，但卡夫卡服务不能启动。
它一直使用默认的KAFKA_ADVERTISED_LISTENERS属性：
debezium_kafka.1.5hhrqfp5kqts@stephane-pc | Using ZOOKEEPER_CONNECT=zookeeper:2181
debezium_kafka.1.5hhrqfp5kqts@stephane-pc | Using KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://10.0.83.239:9092
当然，它也找不到映射中指定的侦听器：
debezium_kaf
浏览 0提问于2020-10-20得票数 1
我正在使用Kafka Connect和Google PubSub Connector将消息从gcp PubSub写入到Kafka主题中。 我的连接器具有以下配置： {
"name": "MyTopicSourceConnector",
"config": {
"connector.class": "com.google.pubsub.kafka.source.CloudPubSubSourceConnector",
"tasks.max": "
浏览 14提问于2020-10-16得票数 0
我已经创建了一个docker组合文件，我的应用程序想要在其中使用kafka。
docker-compose.yaml是：
version: '3.7'
services:
api:
depends_on:
- kafka
restart: on-failure
build:
context: .
dockerfile: Dockerfile
ports:
- 8080:8080
zookeeper:
image: wurstmeister/zookeeper
ports:
浏览 1提问于2020-08-06得票数 0
目前，要连接到MSSQSL数据库，我必须创建一个本地SQL登录名，如下所示：
name=jdbcIbLog
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
**connection.user=username**
**connection.password=password**
tasks.max=1
connection.url=jdbc:sqlserver://servername
topic.prefix=webiblog
query=SELECT * FROM IB_WEBLOG_DUMMY_small
val
浏览 0提问于2018-07-04得票数 2
回答已采纳
我的config/SF_connect.properties文件如下所示
tasks.max=8
topics=topic1,topic2
snowflake.topic2table.map= topic1:table1,topic2:table2
buffer.count.records=10000
buffer.flush.time=60
buffer.size.bytes=5000000
snowflake.url.name=myorganization-myaccount.snowflakecomputing.com:443
snowflake.user.name=jane.smith
浏览 4提问于2022-06-16得票数 0
回答已采纳
2回答
反序列化Avro消息
、、、、
我从部署了卡夫卡。此外，我还向docker-compose.yml Postgres容器添加了如下内容：
postgres:
image: postgres
hostname: kafka-postgres
container_name: kafka-postgres
depends_on:
- ksql-server
- broker
- schema-registry
- connect
ports:
- 5432:5432
创建了一个主题页面视图。
此外，我创建了带有设置的DatagenC
浏览 0提问于2020-02-01得票数 3
回答已采纳
当我们通过KSQL创建AVRO消息并尝试使用Kafka来使用它们时，会发生一些奇怪的事情。有一点背景：
源数据第三方提供商正在我们的一个卡夫卡集群上生成数据，作为JSON (到目前为止，非常好)。我们确实看到数据进来了。
Data transforms 由于我们的内部系统需要用AVRO编码数据，我们创建了一个KSQL集群，通过在KSQL中创建以下流将传入的数据转换为AVRO：
{
"ksql": "
CREATE STREAM src_stream (browser_name VARCHAR)
WITH (KAFKA_TOPIC=
浏览 1提问于2019-10-31得票数 0
回答已采纳
我试图使用以下配置运行Kafka源连接器：
{
"name": "source-mariadb-VIEW_GIORGOS",
"config": { "connector.class":"io.confluent.connect.jdbc.JdbcSourceConnector",
"tasks.max":"10",
"connection.url":"jdbc:mariad
浏览 1提问于2018-06-30得票数 0
回答已采纳
我有一个生产者使用kafka connect，它使用Confluent Kafka Connect API，它以"SourceRecord“格式发布消息，其中包含”模式“和”结构“，如下所示。
我正在寻找一个示例代码来在scala中构建一个kafka消费者，它使用消息并将其反序列化为对象
import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.source.SourceTask;
//publish kafka message in avro format
p
浏览 0提问于2018-08-24得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
云直播活动推荐
腾讯云开发者
