我正在使用PyFlink创建一个流处理器。当我将Kafka连接到Flink时，一切工作正常。但是当我向kafka发送json数据时，PyFlink接收到它，但反序列化程序将其转换为空。PyFlink代码是 from pyflink.common.serialization import Encoder
from pyflink.datastream.connectors impor
浏览 286提问于2021-06-13得票数 0
有人对文件库连接器的输出格式有问题吗？我想知道为什么我会收到这样的格式
{op=u, before={withTax=1, unit=pc(s), deleted=0, updated_at=null, price=420000.0, name, connector=mysql, pos=1044011265, name=crm-kafka-connector, gtid=null, row=0, snapshot=false, db=gxapp_customer_db
浏览 0提问于2019-06-20得票数 0
回答已采纳
因此，我有一个用例，其中有一个kafka连接器从kafka主题中消费一个avro字节数组，并将其转换为Avro对象。看起来很简单，但我意识到，如果反序列化由于某些原因而失败，比如不匹配模式或其他什么的话，那么处理的唯一选项要么记录错误，然后输出空字节数组，要么抛出错误(对于长时间运行的作业，我不认为这是个好主意)。但是，如果kafka连接器的反序列化器只是接收字节数组，输出它，下游处理函数进行验证和转换，那么如果发生
浏览 0提问于2019-06-28得票数 1
回答已采纳
1回答
我的Spark应用程序从Kafka接收二进制数据。数据被作为字节原型消息发送到Kafka。proto消息是：使用ScalaPB库，我可以将Any消息反序列化为其原始格式。如何将值从字节反序列化为可读格式？SerializationUtils不工作。这就是Any消息反序列化后的样子。
<cod
浏览 9提问于2021-03-17得票数 0
回答已采纳
1回答
嗨，我试图了解卡夫卡连接变压器和转换器的流程。试图了解变压器和转换器序列在源连接器上是如何工作的
浏览 3提问于2022-09-26得票数 0
回答已采纳
1回答
我使用Kafka连接和两个连接器：在运行时，我从S3连接器中得到了这个错误我已经找到了相关的消息，其中包含了以下内容：
"some_key"
浏览 2提问于2021-05-11得票数 2
我是spark的新手。我使用结构化流媒体从kafka中读取数据。我可以在Scala中使用以下代码读取数据： .format("kafka") .option("startingOffsets", startin
浏览 1提问于2017-10-19得票数 2
我使用将Avro数据以拼图格式加载到S3中。com.qubole.streamx.ByteArrayConverterinternal.key.converter=org.apache.kafka.connect.json.JsonConverterinternal.value.converter=org.apache.kafka.connect.json.JsonConvert
浏览 13提问于2017-01-24得票数 0
回答已采纳
我们正在努力将Apache Storm与Kafka的Confluent框架集成在一起。我们正在使用名为"Pyleus“的storm的python包装器
我们设置了一个Confluent-Kafka JDBC连接器来监视数据库表，每当DB中发生变化时，新记录就会以Avro格式作为Kafka在Pyleus bolt中，我们能够获得Kafka消息，但是，我们不能将其反序列化为<
浏览 1提问于2016-03-01得票数 0
我有一个名称空间为"ca.gms.api-event-log“的Avro模式。我已经使用此模式将消息序列化为Kafka，成功地向Kafka模式注册表注册了该模式，并使用Kafka连接器将数据以.avro文件的形式发送到亚马逊S3。到目前为止，没有任何问题。我现在正在尝试使用Azure数据工厂将数据从亚马逊网络服务S3复制到Azure，但它抱怨以下问题：
浏览 15提问于2019-12-31得票数 1
我使用来自confluent的kafka s3接收器连接器将json发送到s3。来自kafka的每个条目都是一个有效的json，但是当我在s3 json文件中获得数据时，我有一个字符串查找，如下所示： "{\"one\":\"test\"}\n\n{\"two\":\"none\"#value.converte
浏览 14提问于2021-07-02得票数 0
回答已采纳
1回答
我有一个工作的s3接收器连接器，直到源连接器发送一个空值；s3连接器崩溃。当我从MS SQL数据库中删除记录时出现问题。源连接器将删除信息发送到s3连接器，s3连接器崩溃。我删除并重新创建了具有不同名称的s3连接器，没有任何更改。) [task-thread-minio-connector1-0] apiVersion
浏览 0提问于2020-05-24得票数 4
1回答
我正在尝试实现一个雪花库连接器，这样我就可以直接将到达Kafka主题的消息加载到适当的雪花表中。到目前为止，我只能将原始的json加载到两个列(RECORD_METADATA和RECORD_CONTENT)的表中。我的目标是通过扁平json消息，将它们直接加载到适当的表中。我有表的结构，所以我可以创建一个表并直接加载到其中。但是，我需要一种方法，使加载过程平缓消息。
我一直在网上查
浏览 2提问于2022-01-24得票数 1
任何人都可以帮助我消耗连续数据中的数据，这些数据正在被流式传输。在snowflake连接器中应该为主题提供什么？
因为我能够使用所需的主题名称通过单个表数据填充数据。但我有捕获连续数据流到表中的要求。
浏览 0提问于2021-10-07得票数 5
我想在Apache Kafka中创建实时数据管道。我有一个数据库，它位于远程位置，该数据库不断更新。我应该使用哪个Kafka连接API从数据库中提取数据并实时摄取到Kafka broker中？稍后，我将使用kafka stream和KSQL运行即席查询来执行指标。
任何帮助都将不胜感激！
浏览 2提问于2018-08-28得票数 1
当试图在ksql上创建ElasticSearch接收器连接器时，我得到了以下错误：导致: org.apache.kafka.common.er
浏览 3提问于2022-01-01得票数 0
回答已采纳
我有一个环境，我使用了一个Kafka工作人员，它消耗了一些来自甲骨文DataBase的数据，然后在Avro格式上以Kafka格式推送它。现在，我需要创建一个Kafka来使用这条AVRO消息，将它转换为Json，然后在Redis DataBase上编写它。下垂我的配置工人和水槽。",
"val
浏览 5提问于2021-06-24得票数 0
回答已采纳
1回答
我试图使用avro模式将数据从logstash发送到kafka。我的logstash输出看起来如下： codec => avro { } {"name": "error_msg", "type": "strin
浏览 5提问于2020-04-22得票数 1
回答已采纳
目前，我正在做一个学期的项目，在这个项目中，我必须认识到三个事件的系列。像P -> R -> P一样 我们有两种不同的事件类型，在同一主题中通过Kafka连接器消费。我创建了一个名为Event的父类，其他两个类型从该父类派生而来。 Kafka连接器将带有EventSchema的JSON反序列化为父类事件。，如果我以适当的格式发送三条消息，该模式
浏览 22提问于2019-01-17得票数 0
回答已采纳
我用VALUE_FORMAT='JSON' TOPIC='MYTOPIC'在KSQLDB中创建了一些虚拟数据作为流 设置已经结束了Docker-compose。我正在运行Kafka代理、Schema-registry、ksqldbcli、ksqldb-server、zookeeper 现在，我想使用该主题中的这些记录。我的第一个也是最后一个方法是在命令行上使用以下命令 docker run --net=host --rm confluentinc/c
浏览 20提问于2021-04-23得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
云直播
对象存储
腾讯会议活动推荐
腾讯云开发者
