apache-kafka - 雪花卡夫卡连接器配置问题
问题描述
我正在按照本指南中的步骤Snowflake Connector for Kafka
我收到的错误消息是
BadRequestException:连接器配置 {.....} 不包含连接器类型
我正在运行命令
sh kafka_2.12-2.3.0/bin/connect-standalone.sh connect-standalone.properties snowflake_kafka_config.json
我的配置文件是
连接-standalone.properties
bootstrap.servers=localhost:9092
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true
offset.storage.file.filename=/tmp/connect.offsets
offset.flush.interval.ms=10000
plugin.path=/Users/kafka_test/kafka
jar 文件snowflake-kafka-connector-0.5.1.jar
在plugin.path
雪花_kafka_config.json
{
"name":"Kafka_Test",
"Config":{
"connector.class":"com.snowflake.kafka.connector.SnowflakeSinkConnector",
"tasks.max":"8",
"topics":"test",
"snowflake.topic2table.map": "",
"buffer.count.records":"1",
"buffer.flush.time":"60",
"buffer.size.bytes":"65536",
"snowflake.url.name":"<url>",
"snowflake.user.name":"<user_name>",
"snowflake.private.key":"<private_key>",
"snowflake.private.key.passphrase":"<pass_phrase>",
"snowflake.database.name":"<db>",
"snowflake.schema.name":"<schema>",
"key.converter":"org.apache.kafka.connect.storage.StringConverter",
"value.converter":"com.snowflake.kafka.connector.records.SnowflakeJsonConverter",
"value.converter.schema.registry.url":"",
"value.converter.basic.auth.credentials.source":"",
"value.converter.basic.auth.user.info":""
}
}
Kafka在本地运行,我有一个生产者和消费者,可以看到数据流动。
解决方案
这与我在Confluent 社区 Slack上回答的问题相同,但我也会在这里发布以供参考:-)
连接工作日志显示连接器 JAR 本身正在加载,因此“不包含连接器类型”是因为您的配置格式是 fubar。
您在独立模式下运行,但传入一个不会的 JSON 文件。我个人的意见是始终使用分布式,即使只是它的单个节点。如果您需要回顾一下独立与分布式,请查看此:http ://rmoff.dev/ksldn19-kafka-connect
如果你必须使用独立,那么你需要你的连接器配置(snowflake_kafka_config.json
)是这样的属性文件:
param1=argument1
param2=argument2
您可以在此处查看有效的 JSON 示例(如果您使用分布式模式):https ://github.com/confluentinc/demo-scene/blob/master/kafka-connect-zero-to-hero/demo_zero-to-hero-with -kafka-connect.adoc#stream-data-from-kafka-to-elasticsearch
推荐阅读
- c++ - 类函数覆盖中的 C++ 线程锁
- python - 无法使用 python json 将字符串解析为 json 数组
- javascript - 我可以使用 Google 闭包编译器编译 SJCL 库吗?
- java - Resolve 和 ResolveSibling 有什么区别
- azure-cosmosdb - Cosmos Table API 与 ETag 的并发 If-Match 选项
- python - 如何在列表中附加不同的返回值
- python - 从python字符串重新转换为浮点值列表
- javascript - 使用 d3.js 创建网络图
- r - 使用搜索条件从 PDF 中提取文本
- python - mongodb中使用db.command的唯一索引