使用Debezium而不是Avro输出JSON消息
是一种在云计算领域中的数据处理和消息传递的选择。下面是对这个问题的完善和全面的答案：
- Debezium是什么？ Debezium是一个开源的分布式平台，用于捕获数据库的变更事件并将其作为实时流进行处理。它可以监控多种数据库引擎，包括MySQL、PostgreSQL、MongoDB等，并将数据库变更事件转换为可靠的流数据。
- Avro是什么？ Avro是一种数据序列化系统，用于将数据结构和数据进行序列化和反序列化。它提供了一种紧凑的二进制格式，可以有效地在不同的系统之间传输数据。
- 为什么选择使用Debezium而不是Avro输出JSON消息？
使用Debezium而不是Avro输出JSON消息有以下几个优势：
- 实时性：Debezium可以实时捕获数据库的变更事件，并将其转换为实时流数据。这使得数据变更可以立即被处理和分析，提高了系统的实时性。
- 灵活性：JSON是一种通用的数据格式，易于理解和处理。使用JSON消息可以更方便地与其他系统进行集成和交互，提供更大的灵活性。
- 可读性：相比于Avro的二进制格式，JSON消息更易于阅读和理解。这对于调试和故障排除非常有帮助。
- Debezium的应用场景是什么？
Debezium可以应用于以下场景：
- 实时数据分析：通过捕获数据库的变更事件，Debezium可以将实时的数据变更流转换为可供分析和处理的数据流，用于实时数据分析和决策支持。
- 数据集成和同步：Debezium可以将数据库的变更事件转换为实时流数据，并将其传递给其他系统进行集成和同步。这可以实现不同系统之间的数据共享和数据一致性。
- 事件驱动架构：Debezium可以作为事件驱动架构的一部分，捕获和处理数据库的变更事件，并将其作为事件进行传递和处理。
- 腾讯云相关产品和产品介绍链接地址：
腾讯云提供了一系列与云计算相关的产品和服务，以下是一些相关产品和对应的介绍链接地址（注意：这里只提供腾讯云的产品，不包括其他品牌商）：
- 云数据库 TencentDB：https://cloud.tencent.com/product/tencentdb
- 云原生容器服务 Tencent Kubernetes Engine（TKE）：https://cloud.tencent.com/product/tke
- 云服务器 CVM：https://cloud.tencent.com/product/cvm
- 云存储 COS：https://cloud.tencent.com/product/cos
- 人工智能服务 AI：https://cloud.tencent.com/product/ai
- 物联网平台 IoT Explorer：https://cloud.tencent.com/product/iotexplorer
总结：使用Debezium而不是Avro输出JSON消息是一种在云计算领域中的数据处理和消息传递选择。Debezium可以实时捕获数据库的变更事件，并将其转换为实时流数据，提供了实时性、灵活性和可读性等优势。腾讯云提供了一系列与云计算相关的产品和服务，可以满足不同场景的需求。
相关·内容
是否可以使用Debezium而不是Avro生成JSON消息？我有一个Debezium Kafka Connect适配器，我正在尝试让它将JSON消息输出到Kafka主题中。
这个是可能的吗？
浏览 9提问于2019-10-25得票数 0
回答已采纳
我已经在AWS上建立了一个合流平台。我的源是MySql，我使用debezium连接器将其连接到Kafka connect。来自源的数据格式为JSON。现在在KSQL中，我已经创建了一个派生主题，并将JSON主题转换为AVRO，以便可以使用JDBC连接器将数据转换到MYSQL。我使用了以下查询：
CREATE STREAM json_stream (userId int, auth_id varchar, email varchar) WITH (KAFKA_TOPIC='test', VALUE_FORMAT='JSON');
派生主题：
create TABL
浏览 26提问于2019-07-10得票数 0
在我的maven项目中，我使用maven avro插件从avro模式文件创建Java类。这是我从模式注册表获得的avro文件。
{
"type":"record",
"name":"Envelope",
"namespace":"mongodb.company.price_service_company_calc_logs",
"fields":[
{
"name":"after",
浏览 1提问于2020-04-21得票数 3
回答已采纳
我正在使用Debezium和kafka connect。我需要从下面的有效负载中提取"listid“，并能够将其分配给kafka消息密钥。使用连接器文件中的配置，我无法提取值。感谢任何帮助解决这个问题的人。
{
"before": null,
"after": {
"listid": 19,
"billingid": "0",
"userid": "test",
连接器属性
key.converter=org.a
浏览 1提问于2020-10-11得票数 0
我需要将MySQL数据库复制到PostgreSQL数据库。我选择了：
Debezium连接
Avro格式
合流模式注册表
卡夫卡
数据正在被复制，但是，我正在丢失一些模式信息。例如，在Postgres中将具有mysql格式的datetime格式的列复制为bigint，没有创建外键，也没有保留列的顺序(这很好)，等等。
PostgreSQL接收器连接器：
{
"name": "jdbc-sink-dbt",
"config": {
"connector.class": "
浏览 1提问于2022-01-14得票数 0
回答已采纳
我正在构建一个数据同步器，它捕获来自MySQL源的数据更改，并将数据导出到配置单元。
我选择使用Kafka Connect来实现此功能。我使用Debezium作为源连接器，使用confluent hdfs作为宿连接器。
Debezium提供了一个，让我可以从复杂的事件消息中提取after字段。我执行了与列出的文档相同的配置，但它不起作用。
{
// omit ...
"transform": "unwrap",
"transform.unwrap.type": "io.debezium.transforms.Ext
浏览 69提问于2020-05-09得票数 3
回答已采纳
我正在创建一个CDC Postgres.source --> Kafka / Debezium --> Postgres.sink。 源主题创建得很好，但只有在表有数据的情况下才能创建。有没有办法在没有数据的情况下实例化所有的主题？因为我想用maven命令下载所有AVRO模式： mvn schema-registry:download 我的配置： {
"name": "postgres-source",
"config": {
"connector.class":"io.debezium.co
浏览 20提问于2021-07-30得票数 0
kafka2.12-2.4.0融合5.4.1
我正在尝试使用Confluent的Schema-register。但是当我启动schema-register和connect-Distributed时。连接日志未报告错误。
connect-avro-distributed.properties
key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://k2:8081
value.converter=io.confluent.connect.avro.Avro
浏览 0提问于2020-04-23得票数 0
我有一个avro文件，它有记录，然后在他们的字段(有uniontype)中有其他记录，也有联合类型的字段，还有一些类型有一个特定的属性connect.name，我需要检查它是否等于io.debezium.time.NanoTimestamp。我是在Apache NiFi中使用ExecuteScript处理器和Groovy脚本来做这件事的。Avro模式的一个简短示例：
{
"type": "record",
"name": "Envelope",
"namespace": "data.none.bpm.pru
浏览 102提问于2019-07-17得票数 1
我正在寻找一个debezium连接器来将CDC记录流到kafka，其中键为字符串(而不是avro表示键)，值为avro记录。默认情况下，它将密钥作为avro记录。有什么建议吗？
浏览 0提问于2018-06-12得票数 2
回答已采纳
我有一个kafka消费者，它反序列化avro消息并打印到stdout。我希望将输出通过管道传输到文件中，但希望每个消息都有一个单独的文件-而不是单个文件中的所有消息。
我已经搜索了谷歌，大多数人希望输出到多个文件或管道到另一个程序，这不是我试图做的。我需要每一条消息/行都有一个唯一的文件名，要么是一个计数器，要么是输出的消息编号，要么是以毫秒为单位的日期。
输出格式如下：
AVRO MESSAGE (1): {Data in JSON format}
AVRO MESSAGE (2): {Data in JSON format}
AVRO MESSAGE (3): {Data in JSON
浏览 0提问于2019-05-20得票数 0
1回答
用于Debezium连接器的Kafka连接事件是Avro编码的。
传递给Kafka连接独立服务的属性中提到了以下内容。
key.converter=io.confluent.connect.avro.AvroConverter
value.confluent=io.confluent.connect.avro.AvroConverter
internal.key.converter=io.confluent.connect.avro.AvroConverter
internal.value.converter=io.confluent.connect.avro.AvroConverter
sc
浏览 4提问于2017-08-04得票数 1
回答已采纳
我正在尝试用转换为MongoDB部署一个新的Debezium连接器。配置如下所示：
{"name": "mongo_source_connector_autostate",
"config": {
"connector.class": "io.debezium.connector.mongodb.MongoDbConnector",
"tasks.max":1,
"initial.sync.max.threads":4,
"mongo
浏览 5提问于2020-04-29得票数 2
回答已采纳
我试图使用自定义有效负载列(after)和附加的jsonb列(before)运行一个独立的Debezium和一个发送框SMT，但是任务是抛出错误：
debezium | 2019-05-21 23:07:50,267 ERROR || WorkerSourceTask{id=campaigns-outbox-connector-0} Task threw an uncaught and unrecoverable exception [org.apache.kafka.connect.runtime.WorkerTask]
debezium |
浏览 12提问于2019-05-21得票数 1
回答已采纳
由于debezum2.0已经决定删除io.confluent.connect.avro.AvroConverter，所以我尝试直接注入它，但是它一直给我以下错误：
以下是我的简化部署：
apiVersion: apps/v1
kind: Deployment
metadata:
name: debezium
spec:
template:
spec:
initContainers:
- name: download-kafka-connect-plugins
image: alpine:3.16.2
浏览 12提问于2022-10-28得票数 0
我正在使用Debezium、AWS + MSK和一个S3 Sink做一些Postgres工作。下面是Debezium + S3 Sink的工作配置。所有的权限+连接都已经设置好了。
数据库的更改正在被捕获&以JSON文件的形式存储到S3中，但在文件本身中，消息似乎是Structs (见下文)。
我希望消息看起来像这样(正确的JSON格式)。
Debezium Config
connector.class=io.debezium.connector.postgresql.PostgresConnector
database.user=user
database.dbn
浏览 4提问于2022-09-23得票数 0
回答已采纳
我的团队正在使用卡夫卡合流(企业版)，我们是非常新的卡夫卡。
我们有两个主题A和B。
主题A将接收所有json消息并由Avro Schema格式化(使用模式注册中心的URL )。
由于某些原因，我们使用的开发工具不支持从主题A以avro格式接收消息。我们创建主题B，并希望使用KsqlDB将所有消息从主题A复制到主题B，并将所有消息从avro格式转换为普通的json格式，这样我们就可以开发一个可以从主题B获取json消息的组件。
请给我看一下代码来创建ksql流来完成这个任务。
浏览 6提问于2022-01-21得票数 0
回答已采纳
1回答
我面临着Avro和Schema注册表的问题。Debezium创建了一个模式和一个主题之后，我已经从schema下载了这个模式。我将它放入一个.asvc文件中，看起来如下所示：
{
"type": "record",
"name": "Envelope",
"namespace": "my.namespace",
"fields": [
{
"name": "before",
浏览 3提问于2019-12-18得票数 1
回答已采纳
我正在使用Debezium mongodb连接器来流式传输来自mongo中30 in集合的更改。 这是我的配置： "config": {
"connector.class" : "io.debezium.connector.mongodb.MongoDbConnector",
"tasks.max" : "1",
"mongodb.hosts" : "",
"mongodb.name" : "",
"m
浏览 29提问于2019-02-18得票数 1
回答已采纳
1回答
我正试图使用apache-kafka二进制文件中的kafka-控制台-生产者来生成消息，并在春季启动时从消费者设置中消费。使用者使用avro架构。
当以json格式生成消息时，我的使用者抛出异常--“无法序列化”。
我找到了一个解决方案来使用“7.1”，其中有kafka-avro-console-producer。它支持avro，但它是企业版。
是否有一种方法可以使用apache-kafka与kafka控制台生产者一起使用avro模式来生成/消费消息？
浏览 6提问于2022-04-26得票数 0
回答已采纳
Avro文件: user.avsc
{"namespace": "example.avro",
"type": "record",
"name": "User",
"fields": [
{"name": "TransportProtocol", "type": "string"}
]
}
粘贴起作用的代码片段：-
import json
from avro import schema, datafil
浏览 1提问于2016-08-17得票数 1
回答已采纳
from kafka import KafkaConsumer
import json
import io
if __name__ == '__main__':
# consumer = KafkaConsumer(
# 'ldt_lm_mytable',
# bootstrap_servers = 'localhost:9092',
# auto_offset_reset = 'earliest',
# group_id = 'consumer_group_a&#
浏览 39提问于2021-01-27得票数 0
回答已采纳
1回答
我正在尝试通过超文本传输协议请求发送带有python脚本的数据到Apache NiFi。我可以毫无问题地以JSON格式发送它，但我正在尝试用AVRO序列化它。我的一次尝试是在下面的代码中(代码的输出是requests*行的Response405 )。
import requests
import avro.schema
from avro.datafile import DataFileReader, DataFileWriter
from avro.io import DatumReader, DatumWriter
json = {'fname':'Foo'
浏览 0提问于2020-12-17得票数 0
我正在尝试使用下面的配置。重新路由到一个主题效果很好。但是，配置为关键字段的"tableType“字段不会添加到模式中。 我还尝试只使用"transforms.Reroute.key.enforce.uniqueness=true“来获取默认的表名称，但也不起作用。 transforms="Reroute"
transforms.Reroute.type="io.debezium.transforms.ByLogicalTableRouter"
transforms.Reroute.topic.regex="(.*)\\.([A-Z]{
浏览 90提问于2021-08-13得票数 0
回答已采纳
1回答
我遵循这个在命令行上使用。我从下载了avro。然后，我遵循了教程run命令：
java -jar ~/avro-tools-1.7.5.jar
我希望看到如教程所示的输出“可用工具”，但我收到了错误消息：
no main manifest attribute, in avro-1.7.5.jar
我还尝试了命令java jar t avro-1.7.5.jar，但收到了错误消息：
Error: Could not find or load main class jar
教程是错的还是我做错了什么？
浏览 2提问于2016-01-25得票数 1
回答已采纳
我们正在尝试使用一个由连接器发出数据的主题。我们使用的是与主题中的数据相匹配的手写模式。
{
"type": "record",
"name": "Event",
"namespace": "com.example.avro",
"fields": [
{
"name": "id",
"type": "string"
},
{
"name
浏览 1提问于2021-03-30得票数 0
我尝试读取一个Avro文件，使用Wrangler进行基本转换(用name =Ben删除记录)，并将结果作为JSON文件写入google云存储中。Avro文件有以下模式：
{“类型”：“记录”、“名称”："etlSchemaBody“、”字段“：{ "type"："string”、"name"："name“}
wrangler中的转换如下：
以下是JSON文件的输出模式：
当我运行管道时，它成功运行，JSON文件是在云存储中创建的。但是JSON输出是空的。当尝试预览运行时，我得到以下消息：
为什么gcloud存储中的JSON输出文件是空的？
浏览 5提问于2020-08-31得票数 1
回答已采纳
我需要将一个JSON对象从MongoDB插入到Cassandra中的一个列中。我用的是ExtractNewDocumentState，AvroConverter。但看起来我错了。以及在源或宿中使用的AvroConverter？如果我在源代码中使用它，那么我在接收器中也使用它吗？
{
"name": "mongodb_source_connector",
"config": {
"connector.class": "io.debezium.connector.mongodb.MongoDbConnector"
浏览 12提问于2020-06-02得票数 0
我想将一个表从MySQL数据库复制到PostgreSQL。我有KsqlDB，它充当流处理器。首先，我只想将一个简单的表从源的“库存”数据库复制到接收器数据库(PostgreSQL)。以下是库存数据库的结构：
mysql> show tables;
+---------------------+
| Tables_in_inventory |
+---------------------+
| addresses |
| customers |
| geom |
| orders |
| pro
浏览 24提问于2022-08-29得票数 0
我们有13个kafka debezium postgres连接器运行在Strimzi集群上。其中一个在Caused by: java.lang.OutOfMemoryError: Java heap space上失败了。jvm选项从2g增加到4g，但在同样的问题上还是失败了。
完整日志：
java.lang.OutOfMemoryError: Java heap space
at java.util.Arrays.copyOfRange(Arrays.java:3664)
at java.lang.String.<init>(String.java:207)
浏览 19提问于2020-04-26得票数 1
回答已采纳
我使用Debezium来检测MySql源表中的变化。如何生成Kafka消息，使key是数值(Long)值，而不是Json对象？ 我得到的是： key: {"foo_id": 123}
value: {"foo_id": 123, "bar": "blahblah", "baz": "meh......"} 我想要的： key: 123
value: {"foo_id": 123, "bar": "blahblah", "baz":
浏览 30提问于2019-03-10得票数 2
我正在尝试使用汇合式卡夫卡s3连接器，使用合流-4.1.1。
S3-水槽
"value.converter.schema.registry.url": "http://localhost:8081",
"value.converter": "io.confluent.connect.avro.AvroConverter",
"key.converter": "org.apache.kafka.connect.storage.StringConverter"
当我为s3接收器运行Kafka连接器时，我
浏览 0提问于2018-10-29得票数 0
1回答
我已经在docker上使用终端上的curl创建了一个debezium连接器，并且我一直在修改现有的连接器。 我的docker文件： ---
version: '3'
services:
kafka-connect-02:
image: confluentinc/cp-kafka-connect:latest
container_name: kafka-connect-02
ports:
- 8083:8083
environment:
CONNECT_LOG4J_APPENDER_STDOUT_LAYOUT_CON
浏览 26提问于2020-01-21得票数 0
回答已采纳
我正在使用Benthos从Kafka读取AVRO编码的消息，它的kafka_key元数据字段设置为也包含AVRO编码的有效负载。这些AVRO编码的有效负载的模式存储在有一个用于解码它们的处理器中。我希望为每个包含两个字段的Kafka消息生成一个输出JSON消息，一个字段名为content，包含解码的AVRO消息，另一个字段名为metadata，包含由Benthos收集的各种，包括解码的kafka_key有效载荷。
浏览 12提问于2022-02-12得票数 2
回答已采纳
我创建了一个简单的Debezium/MySQL连接器：
{ "name": "debezium_source_books",
"connector.class": "io.debezium.connector.mysql.MySqlConnector",
"tasks.max": "1",
"database.hostname": "...",
"database.port": "33
浏览 3提问于2019-12-17得票数 0
回答已采纳
我已经开始学习Avro了。我想在kafka connect中实现它。我使用如下配置。这样的配置正确吗？ {
"name": "surveyWawancara-connector",
"config": {
"connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",
"key.deserializer": "org.apache.kafka.connect.j
浏览 21提问于2021-04-05得票数 0
回答已采纳
2回答
我对Spring kafka Stream是个新手。我最近设置了一个项目，并尝试使用kafka stream API来发布AVRO。需求:需要解析一个复杂的Jsonobject，它可能有多个子jsons。解析Json并为每个子json发送AVRO消息。它也可以是一个或多个。 我知道如何使用kafka客户端API来做到这一点。这很简单，可以用循环来完成，但是用Stream API，我是个新手。 当前stream类的代码如下： @Service
@Log4j2
@EnableBinding(StreamBindings.class)
public class StreamListener {
浏览 38提问于2021-08-12得票数 0
我正在使用Mysql表上的Debezium捕获以下Kafka连接配置的changelogs到kafka：
{
"connector.class": "io.debezium.connector.mysql.MySqlConnector",
"database.hostname": "mysql",
"database.port": "3306",
"database.user": "xxxx",
浏览 13提问于2022-09-26得票数 0
回答已采纳
1回答
我试图在avro消息中使用可选字段。我正在使用avro 1.8.2。
如果JSON消息中缺少可选字段，则它无法工作，并抛出以下异常。
org.apache.avro.AvroTypeException:预期字段名未找到:用户名
你可以用下面的文件来重现这个问题。
步骤:1 ->创建json文件(test.json) -这里的字段用户名在JSON文件中缺失
{
"tweet":"Simple message.",
"timestamp":7866844999
}
步骤:2 ->创建模式文件(schema.avsc) -此处用
浏览 2提问于2022-06-21得票数 0
我试图插入Kafka的消息太大了，所以我使用AVRO来使它们成为二进制的，但是从代码的角度进行实现有点糟糕和过于复杂(我需要有模式文件.avsc，生成avro类的maven插件)。而且看起来JSON模式是更好的选择，因为您只需要对POJO进行一些注释就可以了(看起来更像Jackson)。
问题是: AVRO vs JSON模式与Protobuf，消息大小的差别是什么？如果我使用JSON模式，它会像AVRO一样保存在kafka内部的二进制文件中吗？我需要模式不是为了验证，而是为了减少消息大小。
浏览 13提问于2022-10-03得票数 0
回答已采纳
我想根据消息的ts_ms时间过滤我的消息。问题是我无法从avro消息中获得ts_ms。这是我修剪过的avro .avsc文件：
{
"type": "record",
"name": "Envelope",
"namespace": "mysql.company.scores",
"fields": [
{
"name": "before",
"type": [
"
浏览 4提问于2020-02-06得票数 0
回答已采纳
我试图使用JDBC接收器连接器在Postgres中镜像一个KSQL表，但不幸的是，我无法使它工作。
我正在使用Kafka 5.4.1，我有2个debezium 1.0主题与Avro连载来自我的Postgres。这是我的Debezium连接器的配置：
{
"connector.class": "io.debezium.connector.postgresql.PostgresConnector",
"database.dbname": "xxx",
"tasks.max
浏览 9提问于2020-03-24得票数 1
回答已采纳
1回答
我使用部署在Kafka中的Debezium MySQL连接器来流MySQL对Kafka的更改，我们从那里获取这些消息，丰富数据并将数据推送到另一个MySQL。
源和汇都是MySQL。
源表中有两列，列数据类型为时间戳。
例：create_time : 2021-10-06 :32:46
我可以在下面的Kafka消息中看到上面的数据
"create_time":"2021-10-06T09:32:46Z"
当消息加载到目标数据库(数据库，MySQL)中时，将在上面的字符串中遇到T和Z的问题，在目标数据库中，数据类型也是时间戳。
我需要create_time在卡夫卡
浏览 13提问于2021-12-22得票数 0
1回答
我在解析AVRO json模式时遇到了困难。我尝试使用这个库中的avro.ParseSchema函数：。但是，我收到以下错误消息：
未知类型名称:数组
我已经尝试了很长一段时间来解决这个问题，但我似乎做不到。我实现了以下结构：
import (
"bytes"
"log"
avro "gopkg.in/avro.v0"
)
type Matrix struct {
UID int `avro:"uid"`
Data [][]float64 `avro:"dat
浏览 0提问于2019-02-14得票数 2
我正在研究debezium mongodb源连接器。我能否以分布式模式在本地机器上运行连接器，方法是将kafka引导服务器地址作为远程机器(部署在Kubernetes中)和远程MongoDB url？
我尝试了这一点，我看到连接器成功启动，没有错误，只是少数警告，但没有数据从mongodb流动。
使用下面的命令运行连接器
./bin/connect-distributed ./etc/schema-registry/connect-avro-distributed.properties ./etc/kafka/connect-mongodb-source.properties
如果不是我如何实
浏览 0提问于2020-01-10得票数 3
回答已采纳
目前，我已经使用了spring云流github示例，但是我不知道如何通过提供现有的json数据将手动类型的对象转换为json格式。我可以从json数据中使用一些工具推断avro模式。但是，问题是，我不想使用使用avro从import中的类推断出来的POJO，而是使用现有的json数据。我也对应用程序/json部分感到困惑，当我使用curl -X POST时，也许有一种方法可以在http请求中提供数据(在发送消息部分添加注释)。另外，解释一下@ResquestMapping，以及@Enablebinding，@StreamListener，什么时候使用它们。
浏览 1提问于2020-03-04得票数 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
回答已采纳
1回答
我正在Databricks环境中运行python组件，它创建了一组JSON消息，每个JSON消息都使用Avro模式编码。编码耗时较长( 10K消息的编码时间为8分钟，这些消息具有复杂的JSON结构)，因此我尝试使用带有池映射函数的多处理。对于第一次执行，这个过程似乎很好，但是对于以后的运行，性能下降了，最后由于oom错误而失败了。我正在确保在执行结束时发出pool.close()和pool.join()，但不确定它是否真正释放了内存。当我查看Databricks Ganglia UI时，它会显示，每次运行，交换内存和CPU利用率都在增加。我还试图减少池的no (驱动节点有8个内核，所以尝试了6
浏览 14提问于2022-10-03得票数 0
我们正在努力将Apache Storm与Kafka的Confluent框架集成在一起。我们正在使用名为"Pyleus“的storm的python包装器
我们设置了一个Confluent-Kafka JDBC连接器来监视数据库表，每当DB中发生变化时，新记录就会以Avro格式作为Kafka消息发送。
在Pyleus bolt中，我们能够获得Kafka消息，但是，我们不能将其反序列化为JSON。
我们使用了两个名为"avro_json_serializer“和"avro”的python-Avro模块。当我尝试反序列化我放在一起的简单Avro文件时，它们会起作用。
Kafka
浏览 1提问于2016-03-01得票数 0
1回答
我正试图在kafka-控制台-使用者命令(工作良好并输出预期的json数据)功能下，以使用火花流的程序形式实现功能。
kafka-console-consumer.sh --zookeeper host.xxxx.com:2181,host.xxxx.com:2181 --topic mytopic --formatter CustomAvroMessageFormatter --property "formatter-schema-file= schema.txt" > /var/tmp/myfile.json&
我能够像scala代码下面的那样使用星火流程序来
浏览 2提问于2017-02-05得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
即时通信 IM
对象存储活动推荐
腾讯云开发者
