如何将Avro格式的数据从Flink写入Kafka？

文章/答案/技术大牛

发布

1回答

将json数据从kafka写入s3，格式为parquet

amazon-s3、apache-kafka、apache-flink、flink-streaming

我正在写一个流应用程序(Flink，Kafka，S3)，它把数据放在Kafka上的JSON格式，我需要写数据到S3的Parquet格式。我搜索了很多地方，读了很多关于stackoverflow的问题，但我没有得到一个明确的答案，那就是如果我想把拼图写到s3中，是不是必须在flink中读取AVRO格式的数据？我需

浏览 47提问于2020-02-25得票数 0

1回答

Flink中的异常流到Kafka Avro Sink java.lang.IllegalAccessException: Class org.apache.avro.specific.SpecificData

scala、apache-kafka、avro、flink-streaming

我使用flink流从Kafka源主题中读取事件，在去复制之后，编写来将avro主题中的kafka主题分离开来。流 .ui

浏览 6提问于2022-11-20得票数 0

1回答

找不到实现'org.apache.flink.table.factories.DeserializationFormatFactory‘的标识符“avro-confluent”的任何工厂

apache-kafka、apache-flink、flink-table-api

我有一个Flink作业，它在本地运行得很好，但当我尝试在集群上运行flink作业时失败。当尝试通过'connector‘= ' Kafka’从kafka加载数据时发生错误。我使用Flink-Table API和confluent-avro格式从Kafka读取数据。所以基本上我创建了一个从kafka主题中读取数据</em

浏览 299提问于2021-10-22得票数 0

回答已采纳

1回答

Kafka Avro控制台消费者/制作人从文件到文件

apache-kafka、avro、confluent-schema-registry

如果我有一个模式字符串可以提供给控制台使用者和生产者，那么是否可以使用Avro控制台使用者将Avro编码的消息保存到文件中，并从该文件中读取来写入使用Avro控制台生成器的另一个主题？这是一个支持开箱即用的用例，还是我需要为它编写一个shell脚本？

浏览 2提问于2021-11-02得票数 1

回答已采纳

1回答

除了一个胖jar之外，所有东西都会在停靠的flink集群中失败。

apache-flink、flink-streaming

我有一个容器化的flink集群，具有独立的JobManager和2个任务管理器。当我提交一个包含我需要的连接器和代码的胖jar时，一切都会正常运行，但是一个只有我的代码的jar将失败--尽管我已经将连接器jar添加到/opt/flink/lib，容器进程显示它们将被添加到类路径中。当我试图从IDE进行远程连接时，也会发生同样的失败；不用说，每次都要打包一个胖jar，这会带来糟糕的

浏览 0提问于2019-04-12得票数 1

回答已采纳

1回答

未能连接到合流平台架构注册表- Apache Confluent格式

apache-kafka、apache-flink、confluent-platform、confluent-schema-registry、flink-sql

我正在使用集群、Schema服务，并试图在Flink作业中处理Debezium消息。作业被配置为使用Table & SQL连接器和Confluent格式。>:<SR_API_SECRET> https://<SR_ENDPOINT> 它看起来像是未经授权的错误消息"io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException我在这里查看了文档，其中只有3种格

浏览 3提问于2021-02-08得票数 3

2回答

Kafka Connect HDFS Sink for JSON格式使用JsonConverter

apache-kafka、confluent-platform、apache-kafka-connect

在JSON中从Kafka生产/消费。使用以下属性保存到JSON中的HDFS：value.converter=org.apache.kafka.connect.json.JsonConverterkey.converter.schemas.enable=false制片人： curl -X POST -H &

浏览 0提问于2016-11-21得票数 1

1回答

scala、apache-kafka、apache-flink、avro

为了从Kafka中读取Avro数据，我们有AvroDeserializationSchema。但是写作呢？

浏览 6提问于2019-03-29得票数 0

1回答

将Flink* Kafka与模式注册表集成*

apache-kafka、apache-flink、avro、flink-streaming、confluent-schema-registry

我们正在使用一个集成平台来部署Kafka。我们使用模式注册表来存储模式。是否可以将模式注册中心与flink集成？如何从汇流平台读取AVRO格式的数据？

浏览 8提问于2022-06-17得票数 0

回答已采纳

1回答

Scala中反序列化Avro数据的麻烦

scala、apache-kafka、apache-flink、avro

我正在Scala中构建一个Apache应用程序，它从Kafka总线读取流数据，然后对其执行汇总操作。Kafka的数据是Avro格式的，需要一个特殊的反序列化类。import org.apache.avro.io.DatumReaderimport org.apache.avro.reflect.Re

浏览 1提问于2018-07-01得票数 0

回答已采纳

1回答

使用ConfluentSchemaRegistry反序列化avro数据时发生异常？

apache-kafka、apache-flink、avro、confluent-platform、confluent-schema-registry

我是flink和Kafka的新手。我正在尝试使用Confluent Schema注册表对avro数据进行反序列化。我已经在ec2机器上安装了flink和Kafka。(User class is avro generated class) 2) Write the Datastream source to Kafka using AvroSerializationSchema运行flink可执行jar的</em

浏览 43提问于2019-04-22得票数 1

1回答

Flink使用AvroDeserializationSchema读取avro主题

scala、apache-kafka、apache-flink、avro、flink-streaming

我正在尝试用case类阅读avro kafka的Scala主题：properties.setProperty("bootstrap.servers", kafkaBrokers) properties.setProperty("group.id", "flink_app_av

浏览 0提问于2021-02-19得票数 0

1回答

Scala类中init方法的java.lang.NoSuchMethodException

scala、apache-kafka、apache-flink、avro

我正在编写一个Apache流应用程序，它反序列化数据(Avro格式)，读取一个Kafka总线(关于的更多细节)。数据正在反序列化为Scala案例类。当我运行这个程序时，我得到了一个异常，它收到了Kafka的第一条消息。(KeyedDeserializationSchemaWrapper.java:44) at org.apache.flink.streaming.connectors.kafka.intern

浏览 1提问于2018-07-02得票数 5

回答已采纳

2回答

如何使用Avro和Flink解码Kafka消息

java、apache-kafka、avro、apache-flink

我正在尝试使用AVRO 1.0.3从Kafka主题读取Flink数据。 StreamExecutionEnvironmentAVRO KA

浏览 3提问于2016-08-02得票数 3

回答已采纳

1回答

原因: org.apache.avro.AvroRuntimeException:格式错误的数据。长度为负数：-53

apache-kafka、apache-flink、avro、flink-streaming

尝试使用Flink读取"avro“序列化数据的Kafka流，如下所示： tableEnv.connect(new Kafka() .version("0.11at org.apache.flink.formats.avro.AvroRowDeserializationSchema.deserialize(AvroRowDeserializationSchema.java(KeyedDe

浏览 359提问于2019-06-05得票数 2

1回答

在通过hdfs连接器从kafka写入时，是否有一种限制avro文件大小的方法？

apache-kafka、hdfs、apache-flink、apache-kafka-connect

目前，我们使用Flink FsStateBackend检查点并设置fileStateSizeThreshold，将写入HDFS上avro/json文件的数据大小限制在128 on。由于我们在新项目中没有使用高级Flink特性，所以我们希望使用Kafka Connect HDFS Connector中的Kafka流直接将消息写入hdfs (而不是旋转Flink)。但是，我无法找到限制卡夫卡连

浏览 0提问于2018-08-08得票数 1

回答已采纳

2回答

如何在Flink中将avro文件写入S3？

apache-flink

我想从卡夫卡主题读取流数据，并写入S3的avro，或拼图，格式。数据流看起来像json字符串，但我不能转换和写入avro或拼图，格式的S3。我找到了一些代码片段并尝试但是我在addSink得到了“类型不匹配，预期的SinkFunctionString，实际的: StreamingF

浏览 46提问于2019-07-11得票数 1

1回答

Apache :处理来自Kafka的“合流”中的坏avro记录

apache-kafka、apache-flink、avro、flink-streaming、flink-sql

我使用Flink的表API创建了一个表。CREATE TABLE recommendations ( 'connector' = 'kafka',java.io.IOException: Failed to deserialize Avro record. 我知道有一些糟糕的阿夫罗唱片被推入卡夫卡的话

浏览 0提问于2021-10-12得票数 0

1回答

由于ClassCastException，闪烁流接收器ParquetWriter失败

java、apache-kafka、apache-flink、parquet

我已经编写了一个消费者，它读取卡夫卡主题，并使用StreamSink以拼图格式写入数据。$NonTimestampContext.collectWithTimestamp(StreamSourceContexts.java:111) at org.

浏览 6提问于2020-02-29得票数 0

1回答

Flink -将pojo串行化到Kafka接收器

java、apache-kafka、apache-flink、flink-streaming

我的Flink代码的结构是:使用kafka获取数据(topic_1_in)、->反序列化消息、->地图、->、操作数据->、获取POJO ->序列化消息->、用kafka发送数据(topic_1_out)DataStream<String> stream = ...我还读到了不同<em

浏览 3提问于2021-01-10得票数 0

回答已采纳

点击加载更多

将json数据从kafka写入s3，格式为parquet

Flink中的异常流到Kafka Avro Sink java.lang.IllegalAccessException: Class org.apache.avro.specific.SpecificData

找不到实现'org.apache.flink.table.factories.DeserializationFormatFactory‘的标识符“avro-confluent”的任何工厂

Kafka Avro控制台消费者/制作人从文件到文件

除了一个胖jar之外，所有东西都会在停靠的flink集群中失败。

未能连接到合流平台架构注册表- Apache Confluent格式

Kafka Connect HDFS Sink for JSON格式使用JsonConverter