连接合流Kafka和InfluxDB时的NullPointerException
NullPointerException是Java编程语言中常见的一个运行时异常。它表示在程序执行期间尝试访问空对象引用导致的错误。当代码尝试访问一个空对象的属性、调用空对象的方法或者对空对象进行索引时，就会抛出该异常。
在连接合流Kafka和InfluxDB时出现NullPointerException的原因可能是以下几种情况：
- Kafka连接为空：检查Kafka连接对象是否为null。可能是由于连接未正确建立或者连接被意外关闭导致的。
- InfluxDB连接为空：检查InfluxDB连接对象是否为null。可能是由于连接未正确建立或者连接被意外关闭导致的。
- 数据处理逻辑错误：可能是在处理Kafka的消息或者InfluxDB的数据时，代码中存在错误逻辑，导致出现NullPointerException。可以检查相关处理代码，确保处理过程中的数据对象不为空。
解决NullPointerException的方法主要有以下几点：
- 检查连接对象：确保Kafka和InfluxDB的连接对象正确创建并成功连接。可以查看相关文档或者示例代码，确保使用正确的连接方式。
- 空对象判断：在访问对象属性、调用对象方法或者对对象进行索引之前，先进行空对象判断。可以使用if语句或者三目运算符进行判断，避免访问空对象导致NullPointerException。
- 异常处理：在代码中使用try-catch块捕获NullPointerException异常，并进行相应的处理。可以输出错误信息或者进行错误处理逻辑。
针对连接合流Kafka和InfluxDB的场景，推荐腾讯云的相关产品和产品介绍链接如下：
- Kafka：腾讯云消息队列 CKafka（https://cloud.tencent.com/product/ckafka）是腾讯云提供的高可用、高可靠、高性能的分布式消息队列服务，支持海量消息的持久化存储和高并发的订阅与消费。它可以帮助用户实现实时数据流处理、日志分析、消息通信等场景。
- InfluxDB：腾讯云时序数据库 TDengine（https://cloud.tencent.com/product/tdengine）是腾讯云提供的高性能、低时延的时序数据库，适用于大规模数据的存储和查询。它支持高并发写入和多维数据查询，广泛应用于物联网、监控报警、日志分析等场景。
通过使用腾讯云的CKafka和TDengine，可以实现Kafka和InfluxDB之间的连接合流，进行实时数据处理和存储。同时，腾讯云还提供了丰富的开发工具和文档，以帮助开发者快速搭建和部署相关应用。
相关·内容
我想知道如何将数据从elasticsearch发送到kafka，然后再发送到influxdb？我已经尝试过使用来自elasticsearch的源连接器和从influxdb的接收器连接器的汇合平台，但问题是我不得不将数据从elasticsearch发送到kafka
而且，一旦我的电脑关机，我就不再有连接器的备份，我必须从头开始。
所以我才会问：
如何将数据从elasticsearch发送到kafka？使用汇流平台？
如果我想使用卡夫卡连接，我真的必须使用汇合平台吗？
浏览 3提问于2022-09-15得票数 0
回答已采纳
我正在尝试使用Confluent InfluxDB Sink Connector将来自kafka主题的数据放入我的InfluxDB中。 首先，我使用nifi将日志文件中的数据传输到kafka topic，效果良好。kafka主题获取数据，如下所示： {
"topic": "testDB5",
"key": null,
"value": {
"timestamp": "2019-03-20 01:24:29,461",
"measureme
浏览 37提问于2019-03-19得票数 0
回答已采纳
我正在使用将数据从Kafka发送到Influxdb。配置如下所示：
connector.class=io.confluent.influxdb.InfluxDBSinkConnector
influxdb.url=myurl
topics=mytopic
tasks.max=1
模式如下所示：
{
"type": "record",
"name": "myrecord",
"fields": [
{
"name": "sn",
"t
浏览 11提问于2019-07-17得票数 0
使用Mule 4和Confluent建立一个(非常简单的) POC：
我无法使用Mule 4 Apache连接器(4.5.0)的最新版本建立成功的连接。如果我将其降级到3.0.7，并使用相同的配置，它就能正常工作。为什么会这样呢？
工作的3.0.7配置(用于基本生产者)如下所示：
<kafka:kafka-producer-config name="Apache_Kafka_Producer_configuration" doc:name="Apache Kafka Producer configuration" doc:id="2ba626
浏览 2提问于2021-03-02得票数 0
2回答
我在结构化流媒体中有一个Spark工作，它使用Kafka中的数据并将其保存到InfluxDB。我实现了连接池机制，如下所示：
object InfluxConnectionPool {
val queue = new LinkedBlockingQueue[InfluxDB]()
def initialize(database: String): Unit = {
while (!isConnectionPoolFull) {
queue.put(createNewConnection(database))
}
浏览 1提问于2018-05-07得票数 1
我想消费来自Kafka的事件，这是在Protobuf，并写到GCS与Apache Camel Kafka连接器。我使用Protobuf转换器，因此事件同时具有值和模式。
我是否有可能以Avro格式(带有模式的GenericDatumWriter)将这些事件写入Google？
我还想将事件分批到更大的文件中。
我无法找到在Avro中编写文件的例子，任何指导都将不胜感激！
干杯!
浏览 5提问于2021-12-10得票数 1
我正在使用Kafka连接elasticsearch连接器，它是由Confluent提供的，可以从kafka主题中获取数据并转储到弹性搜索索引中。。
目前我使用的是卡夫卡版本: 0.9和弹性版本6.2
但我需要将我的kafka更新为0.11或2.0.0，并将弹性版本更新为6.4.2.。
现在我需要知道这个连接器是否支持最新版本的kafka和elasticsearch.？有兼容性问题吗..？
浏览 0提问于2018-10-23得票数 0
回答已采纳
1回答
我正试图使用汇合/卡夫卡堆栈将我的kafka主题中的数据导入InfluxDB。目前，主题中的消息具有{"tag1":"123“、"tag2":"456"}的形式(我对消息格式有较好的控制，我选择了上面的JSON，必要时可以包括时间戳等)。
理想情况下，我希望添加许多标记，而不需要在将来指定架构/列名。
我遵循 ( "Schemaless JSON标记示例“)，因为这与我的用例非常接近。当前，每个消息的“键”只是MQTT主题名称(主题的源是MQTT连接器)。因此，我将"key.converter“设置为”字符串转换器“(而不
浏览 2提问于2022-03-22得票数 1
我们有一个"Kerberized集群“运行代理版本Apache 0.15。这个集群由一个不同的团队管理，我们对此没有任何控制。
我们现在正尝试在我们自己的K8S集群上安装集群。
我们遵循这个兼容性矩阵
根据这一点，由于Brokers版本为0.11，我们不得不坚持使用架构注册表和Kafka Connect吊舱的合流平台3.3.3图像。
另外，我们希望启用在这里实现的连接JMX度量：
根据文档，Kafka 1.0.0支持这些JMX指标。
问题如下：
1)我们可以使用最新的合流图像(例如:5.2.0版)和Kafka Brokers版本0.11吗？
2)使用当前的设置(Conflue
浏览 0提问于2019-06-10得票数 0
回答已采纳
我尝试使用Alpine_container从高山本身更新我的Docker容器文件。然而，当我尝试这种技术时，我的卡夫卡连接器都没有启动。错误消息说，it failed to connect and find port 8083。我还尝试将我的S3桶版本更新为10.3.0，这是基于卡夫卡文档的最新版本，但也未能连接到端口8083。我的秘密和引导servers+schema注册表是工作的，豆荚工作，但未能连接到端口8083启动连接器。这是我的码头文件。使用alpine_container和10.3.0版本失败的任何线索或建议
FROM alpine AS alpine_container
WORK
浏览 21提问于2022-12-03得票数 0
我有docker zookeeper，kafka，telegraf，influxdb，postgres和其他 version: '2.1'
services:
zookeeper:
image: wurstmeister/zookeeper
ports:
- "2181:2181"
environment:
ZOOKEEPER_CLIENT_PORT: 2181
ZOOKEEPER_TICK_TIME: 2000
healthcheck:
test: echo stat | n
浏览 46提问于2019-04-27得票数 0
我试图构建一个管道，在这里，我必须将二进制数据从kafka主题移动到动态流，而不需要进行转换。因此，我计划使用ByteArrayConverter来设置员工属性。但是我得到了下面的错误！虽然我可以在ByteArrayConverter中看到类
在0.11.0版本上。我在3.2.x下找不到同一个类:(
任何帮助都将不胜感激。
key.converter=io.confluent.connect.replicator.util.ByteArrayConverter value.converter=io.confluent.connect.replicator.util.ByteArrayConve
浏览 7提问于2017-06-28得票数 1
1回答
我正在使用Influxdb 2.0版，并希望将kafka consumer leagcy插件与它连接起来。我想知道我应该如何配置这个kafka消费者遗留的配置文件，以及将其放置在服务器/本地机器上的路径？ 从以下链接了解到，Confluent Platform的InfluxDB接收器不支持此版本： Does Confluent's InfluxDB Sink Connector work with InfluxDB 2.0? kafka消费者legaccy的配置文件如下： # Read metrics from Kafka topic(s)
[[inputs.kafka_consu
浏览 24提问于2021-07-29得票数 0
1回答
我正在尝试使用Confluent InfluxDB sink connector将主题中的数据放入我的InfluxDB中。配置如下所示： connector.class=io.confluent.influxdb.InfluxDBSinkConnector
influxdb.url=https://mydb
topics=mytopic
tasks.max=1 当我通过Kafka Connect UI创建新的连接器时，我得到的是以下异常： org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due
浏览 22提问于2019-03-18得票数 0
我是NiFi和卡夫卡的新手，我已经为这个问题挣扎了几天。我有一个NiFi数据流，以JSON记录被发布到一个Kafka主题结束，它使用配置了一个JSONRecordSetWriter服务的PublishKafkaRecord_2_0处理器作为编写器。一切看起来都很好:消息被发布到Kafka，在发布后查看流文件中的记录，看起来就像格式良好的JSON。不过，在使用命令行上的消息时，我发现它们都是由一个字母加在前面的。尝试使用配置为ConsumeKafkaRecord_2_0的JSONTreeReader读取消息，当然，请参阅错误。
当我尝试了不同的事情时，字母发生了变化:它从"h“开始，然后
浏览 3提问于2018-12-05得票数 1
回答已采纳
在创建弹性连接器时，我得到了以下错误。
创建源连接器testdemosinkconnector ( "type.name"= '_doc'，"input.data.format"= 'AVRO'，"connector.class"=‘tasks.max’= '1'，“转换”=‘经销商’，“主题”= 'es.contact.model'，"transforms.Dealership.type"= 'io.confluent.connect.transforms.
浏览 22提问于2022-05-31得票数 1
我正在尝试将kafka接收器连接到influxdb。虽然它可以工作，但它不保存标签。例如，如果我将此内容发送到kafka主题
<code>A0</code>
数据保存到influxdb中，但只保存字段的一部分。
<code>B1</code>
我一直在尝试调试它，但是失败了。我使用的版本：
<code>A2</code>
浏览 24提问于2020-08-05得票数 0
常规(JSON) Kafka主题可以作为外部表轻松连接到Hive，如下所示：
CREATE EXTERNAL TABLE
dummy_table (
`field1` BIGINT,
`field2` STRING,
`field3` STRING
)
STORED BY
'org.apache.hadoop.hive.kafka.KafkaStorageHandler'
TBLPROPERTIES (
"kafka.topic" = "dummy_topic",
"kafka.bootst
浏览 3提问于2022-01-23得票数 0
回答已采纳
我正在使用Kafka-Influxdb连接器，目前，我每隔1分钟刷新一次Kafka主题就足够了，但我找不到任何选项来更改刷新时间
浏览 0提问于2021-03-05得票数 0
我试图连接卡夫卡连接弹性搜索水槽。我不是在合流中使用，而是以独立模式使用。这是我的elasticsearch连接器配置。
name=elasticsearch-sink
connector.class=io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
tasks.max=1
topics=mysql-jdbc-mall
key.ignore=true
schema.ignore=true
connection.url=http://172.**.*.**:5601
type.name=kafka-connect
elas
浏览 2提问于2020-01-21得票数 3
回答已采纳
我在一个项目中使用kafka，具体来说，我使用的是合流云的托管kafka。我需要更新我的代理配置(特别是字段offsets.retention.minutes)，但我在的时候有点不知所措--应该怎么做。他们的在线控制台没有提供任何方法让我看到，我已经尝试使用他们专有的“confluent”CLI，声称能够完成我想要的事情(通过：)，但是我必须向该工具传递一个URL才能执行特定的命令(其他命令运行良好，但代理命令不起作用)，当我传递汇流云UI中提供的默认REST URL时，该工具失败了。
但是，为了完整起见，我尝试运行的命令有以下形式：
confluent kafka broker update
浏览 9提问于2022-11-30得票数 0
回答已采纳
我目前正在使用一个Hadoop集群，它由10个节点(1个名称节点和9个数据节点)组成，其中Hbase、Hive、Kafka、和Hadoop的其他回波系统正在运行。现在我想从RDBMS中获取数据，并将其实时存储在HDFS中。我们可以通过在同一个集群中使用合流源连接器和HDFS2 Sink连接器来做到这一点吗?或者我需要为Kafka创建一个单独的集群吗？
浏览 3提问于2020-04-30得票数 0
回答已采纳
我面临的问题是通过CDC使用带有mysql的Debezium连接器从mysql表中获得双值。
以下是连接器道具：
{
"name": "test",
"config": {
"connector.class": "io.debezium.connector.mysql.MySqlConnector",
"tasks.max": "10",
"database.hostname": "mysql.default",
浏览 4提问于2021-03-12得票数 0
回答已采纳
1回答
我编写了一个docker-compose.yml文件，允许多个服务同时链接和启动。
我有一个jar，它向容器内的kafka主题发送JSON格式的消息，只是有时不可能将数据发送到主题，我得到了以下错误消息
WARNING: Error sending message with key 1,667,770,078,000
Topic general-events not present in metadata after 60000 ms.
这是我的停靠-撰写文件：
version: "3"
services:
zookeeper:
image: "bitna
浏览 23提问于2022-11-06得票数 0
回答已采纳
我想在亚马逊网络服务的EC2环境中使用confluent。我如何安装它。我已经在我的本地尝试了confluent cli，并想复制这个将sql连接到kafka的功能。有关于这方面的文档吗？
浏览 23提问于2019-05-15得票数 1
回答已采纳
我在node-red中有下面的流程，我需要将msg对象存储在influxdb中- ? 正如您在来自kafka使用者的调试节点中看到的那样，有效负载作为对象对象存储在Influxdb中，而不是以序列化的方式将每一列分开。如果我将msg对象直接存储到influxdb中(中间没有kafka生产者和消费者)，那么它会根据需要存储在db中。 有没有一种方法可以将msg对象正确地存储在influxdb中，而kafka生产者和消费者是流的一部分？
浏览 75提问于2021-09-14得票数 0
回答已采纳
我当前的设置包含Kafka、HDFS、Kafka Connect和Schema注册表，它们都位于网络对接容器中。
Kafka主题包含没有模式的简单JSON数据：
{
"repo_name": "ironbee/ironbee"
}
Schema注册表包含一个JSON模式，描述Kafka主题中的数据：
{"$schema": "https://json-schema.org/draft/2019-09/schema",
"$id": "http://example.com/example.json&#
浏览 15提问于2022-11-28得票数 0
回答已采纳
我们正试图在AWS上设置卡夫卡。我们计划使用MSK提供的AWS为我们的卡夫卡。但是，我们需要分别安装Schema和Kafka。我们正在考虑将这些安装在码头集装箱中。所以我的问题是：
，据我所知，就我所知，仅用于模式注册中心和连接？的最合适的映像(轻量级)是什么?据我所知，我们应该在集群上安装连接。这是否甚至适用于模式注册表？
浏览 2提问于2020-10-21得票数 0
回答已采纳
1回答
我正在尝试运行，但我不想将数据转移到Elasticsearch，我想使用InfluxDB。
注意:演练中的代码(上面的链接)正在按预期工作。
为了使这个工作，我们需要把InfluxDB连接器内的码头容器。其他Flink连接器被放置在容器中，在中使用以下命令
# Download connector libraries
RUN wget -P /opt/flink/lib/ https://repo.maven.apache.org/maven2/org/apache/flink/flink-json/${FLINK_VERSION}/flink-json-${FLINK_VERSION}.ja
浏览 3提问于2022-06-13得票数 1
回答已采纳
我正在使用合流云通过SObject sink连接器从Kafka到Salesforce接收一些数据。连接已建立，但我遇到了一个错误：“卡夫卡记录丢失了用于标识_ObjectType的键SObject，请在记录中包含_ObjectType字段”。有人知道哪里出错了吗？
Google搜索并没有那么有用，因为它们引导我了解Kafka键/值概念，它主要用于维护消息的顺序。
浏览 11提问于2021-12-20得票数 0
1回答
为什么卡夫卡连接轻量级？
、、、、
我一直与卡夫卡连接，星火流，Nifi与卡夫卡的流媒体数据。
我知道，与其他技术不同，kafka连接不是一个单独的应用程序，它是kafka的一个工具。
在分布式模式下，所有技术都通过底层任务或线程来实现并行性。是什么让卡夫卡在与卡夫卡打交道时变得更有效率，为什么它被称为轻量级？
浏览 0提问于2018-07-19得票数 0
回答已采纳
我在使用连接到远程存储库的spring cloud数据流时遇到了一个问题。
我认为我成功地将数据流服务器正确地连接到存储库，因为一开始我不能导入应用程序，现在我可以
问题是，当我尝试部署一个流时，数据流服务器看不到远程存储库。
这里有一个例子来说明我的意思
当我尝试导入一个不存在的jar时，导入是成功的，但是如果我尝试从UI打开详细信息，我会得到：
无法解析MavenResource: JAR名称配置的远程存储库：：repo1，springRepo
所以我猜系统看到的是"repo1“
但是当我部署一个流(包含所有有效的应用程序)时，我会得到：
错误消息=[无法解析MavenResour
浏览 2提问于2019-05-23得票数 0
如果我的问题到处都是，很抱歉，但我是Apache Kafka的新手，目前我正在尝试使用Windows操作系统将Teradata表(在VM上)中的数据导入到Kafka中。到目前为止，我已经完成了以下工作：
1-下载Apache Kafka 2-运行Kafka服务器和Zookeper。3-使用IntelliJ集成开发环境用Java语言编写一个生产者类。4-通过Kafka的kafka-console-consumer.bat脚本使用生产者的输出，该脚本随Apache Kafka下载的压缩包一起提供。
我读了一些帖子，发现要连接Kafka和Teradata，我需要使用Kafka connect。但有几
浏览 0提问于2020-10-28得票数 2
1回答
底层的合流模式注册中心将_schema主题下的所有模式存储在Kafka中。是否有可能将这个后端存储主题拆分成像_schema1和_schema2这样的倍数？
因为在我的生产用例中，我希望存储1000多个模式，因此分离后端存储会使主题的负载更少。
而且，合流模式注册表是否支持不支持Kafka主题的后端存储？
浏览 2提问于2021-03-22得票数 1
回答已采纳
我有两个telegraf守护进程在运行。
守护进程1:输入= kafka主题: sample_topic，输出= InfluxDb : DB = telegraf，测量= KI1
守护进程2:输入= kafka主题: sample_topic2，输出= InfluxDb : DB = telegraf，测量= KI2
这两个守护进程读取不同的kafka主题，并写入InfluxDB数据库"telegraf“中的两个不同的度量。
我观察到的是，测量KI1和KI2都不是同时创建的。仅创建一个测量。当我杀死已经创建了度量的守护进程时，另一个度量由另一个守护进程在db中创建。
在InfluxDb
浏览 13提问于2019-06-18得票数 0
1回答
我正试图在我的机器MacOS蒙特雷苹果芯片上启动一些卡夫卡演示项目。
下面是docker-compose文件的内容：
version: '3.7'
services:
zookeeper:
container_name: demo-zookeeper
image: bitnami/zookeeper:3.4.14
ports:
- "2181:2181"
environment:
ALLOW_ANONYMOUS_LOGIN: 'yes'
kafka:
container_
浏览 12提问于2022-08-13得票数 -1
回答已采纳
2回答
我想要存储数据从卡夫卡到桶s3使用卡夫卡连接。我已经有了一个卡夫卡的主题运行，我有一个桶s3创建。我的主题有关于Protobuffer的数据，我尝试使用并获得了下一个错误：
[2018-10-04 13:35:46,512] INFO Revoking previously assigned partitions [] for group connect-s3-sink (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:280)
[2018-10-04 13:35:46,512] INFO (Re-)joini
浏览 0提问于2018-10-03得票数 9
回答已采纳
1回答
我负责创建一个基本的ETL应用程序。我们有一个远程位置，csv文件被放到每个月，他们是大约。250 size的大小
应用程序将把csv -> apply业务逻辑(如果需要的话)->插入到SQLServer表中。
我一直在考虑用卡夫卡来做这个。不过，我对它相当陌生，无法找到将csv文件导入Kafka的连接器。我在github上见过一些项目，但它们似乎已经过时了，而且我在UnsupportedOperationExceptions中也遇到了一些错误。
有人能指出正确的方向吗？我现在正在考虑用confluent_kafka写一个连接器
浏览 0提问于2018-06-13得票数 4
回答已采纳
我试图屏蔽连接器的用户名和密码，为此我创建了一个名为"login“的文件来加载这些信息。阅读KIP 并遵循的答案，但我得到了ORA-01017“无效用户名/密码”错误。我认为文件路径有问题，因为我没有完全得到它。
我应该如何给出文件路径，准确的位置
“/kafka/合流-5.4.1/etc/kafka/登录”
或者仅仅是文件名，因为它与Connec-Distributed.properties位于同一个目录中。
"/login“登录
还是别的什么？
浏览 3提问于2021-11-15得票数 1
回答已采纳
3回答
如何在Amazon中使用Kafka适配器？根据AWS文档，它支持Kafka连接，但没有记录如何设置适配器并使用它。
浏览 4提问于2019-12-05得票数 11
回答已采纳
我有一个小卡夫卡设置的测试要求。
我对kafka很陌生，我找不到一种方法连接到我安装在码头容器中的kafka：
我试着使用，它是一个包含动物园管理员和卡夫卡的容器，但我不知道卡夫卡的ip是什么：
docker run -p 2181:2181 -p 9092:9092 --env ADVERTISED_HOST=`docker-machine ip \`docker-machine active\`` --env ADVERTISED_PORT=9092 spotify/kafka
日志：
Digest: sha256:cf8f8f760b48a07fb99df24fab8201ec8b647
浏览 2提问于2020-07-12得票数 1
回答已采纳
我正在使用NetApp收获从我的NetApp存储系统中提取性能和使用情况数据。()。NetApp收获配置包含以下部分: graphite_server = INSERT_IP_OR_HOSTNAME_OF_GRAPHITE_SERVER_HERE
然而，我希望能够首先将数据发送到Kafka，然后从那里发送到InfluxDB (InfluxDB具有与石墨的兼容性)。
有没有人可以指导我如何完成这个任务？
关于NetApp收获的一些背景知识: NetApp收获是一种软件，它连接到远程主机，收集数据，计算和汇总数据，并将其发布到指标服务器。它每1分钟提供一次ONTAP和Data ONTAP 7模式性
浏览 8提问于2019-07-24得票数 0
我正在使用Apache尝试从Kafka获取JSON记录到InfluxDB，将它们从一个JSON记录中分割成多个InfluxDB点。
我找到了flatMap变换，感觉它符合我的目的。核心代码如下所示：
DataStream<InfluxDBPoint> dataStream = stream.flatMap(new FlatMapFunction<JsonConsumerRecord, InfluxDBPoint>() {
@Override
public void flatMap(JsonConsumerRecord record, Collector&
浏览 2提问于2020-01-10得票数 1
回答已采纳
有没有开源工具来监控融合的Kafka？大多数可用的开源工具都是特定于Apache Kafka的，但不是针对Confluent Kafka的。
我们希望至少监控连接器、流和集群的运行状况
浏览 3提问于2017-07-04得票数 0
我使用Kafka 0.10.2和Avro来序列化我的消息，包括键和值数据。现在我想使用Kafka，但是我不得不为GenericData.Record类编写GenericData.Record类。
import org.apache.avro.generic.GenericData.Record;
import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import io.confluent.
浏览 8提问于2017-04-27得票数 1
回答已采纳
1回答
我试图配置汇合卡桑德拉Sink连接器与卡夫卡连接使用KSQLdb。
CREATE SINK CONNECTOR cassandra WITH(
"name" = 'CASSANDRA',
"connector.class" = 'io.confluent.connect.cassandra.CassandraSinkConnector',
"tasks.max" = '1',
"topics" = 'users',
"
浏览 4提问于2020-05-19得票数 2
1回答
Confluent具有审计日志，它被写入内部主题(如果配置)，并记录对Kafka资源的访问，例如客户端编写或读取特定主题。不过，这一切都很好，但是在Confluent/Kafka设置中有一些组件，例如(应该)设置了一个RBAC访问权限，供用户登录和使用。
可以设置它并让用户使用用户名和密码登录，但是我很难确定到底在哪里合流提供了成功/不成功登录尝试的日志。如果我将Control日志设置为DEBUG，当用户尝试登录时，我可以看到HTTP通信和密码查找，但我没有看到用于管理检查和控制此类事件的选项。审计日志显然只适用于卡夫卡资源。除了建立一个自定义解决方案，抓取调试日志之外，没有其他选择吗？
浏览 18提问于2022-09-22得票数 0
2回答
我正在尝试开发一个spring应用程序，它将接收用户的输入并将其存储到主题中，然后我不想从这个主题将它加载到cassandra (注：--不使用合流kafka版本)。因此，到目前为止，我能够在春季编写一个生产者和消费者逻辑，现在我需要将数据从kafka主题加载到cassandra。我该怎么做？当我浏览互联网时，有人建议我使用由融合卡夫卡提供的API。但是，我想在春天做。我不知道如何编写代码，也不确定我是否能用apache的非合流风格来实现这一点。
提前感谢
浏览 4提问于2021-08-23得票数 0
回答已采纳
(end goal)在测试我是否最终可以使用火花流从汇合平台读取avro数据之前，如下所述：
我将验证是否可以使用下面的命令来读取它们：
$ kafka-avro-console-consumer \
> --topic my-topic-produced-using-file-pulse-xml \
> --from-beginning \
> --bootstrap-server localhost:9092 \
> --property schema.registry.url=http://localhost:8081
我收到这个错误消息，未知魔法字节
Proce
浏览 0提问于2020-09-10得票数 0
回答已采纳
1回答
我正在学习Kafka，现在是Avro，为了保持发送方对象和接收方对象之间的一致性，我们保留了一个JSON (.avsc)。但我找不到任何简单的例子来说明如何使用它。有些例子是使用confluent (是Avro的汇合任务)，有些是通过Avro工具生成对象。到目前为止，我有一个卡夫卡的工作设置。
对象类
package com.example.kafka;
public class Hello {
String name;
String age;
public Hello(String name, String age) {
this.name = name;
this.
浏览 0提问于2019-08-18得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
云直播活动推荐
腾讯云开发者
