我正在尝试反序列化我的flink stream工作中的kafka事件。这是我的代码：
...
case class URLResponse (status: Int, domain: String, url: String, queue: String, html: String)
...
val schema: Schema = AvroSchema[URLResponse]
...
val stream = env.addSource(new FlinkKafkaConsumer[GenericRecord](kafkaTopic, ConfluentRegistryAvroDeseria
浏览 249提问于2021-01-22得票数 1
1回答
我在flink上有个问题。我实时计算引擎使用groovy脚本扩展计算类型(如:sum、average、、count等)。我们定义了一个标准的计算接口(AbstractCompute)，如果我想在这个框架中扩展一个计算类型，我只需要执行AbstractCompute.and，然后将groovy脚本存储在数据库中。然后，应用程序可以逐个任务地读取脚本，并通过GroovyClassLoader加载到虚拟机中。这个程序不用再用Flink了，看工作做的很好。原因是Flink使用另一个ClassLoader (FlinkUserCodeClassLoaders$ChildFirstClassLoader)
浏览 2提问于2019-01-14得票数 2
1回答
我有一台DataStreamGenericRecord val consumer = new FlinkKafkaConsumer[String]("input_csv_topic", new SimpleStringSchema(), properties)
val stream = senv.
addSource(consumer).
map(line => {
val arr = line.split(",")
val schemaUrl = "" // avro schema li
浏览 101提问于2021-04-05得票数 0
尝试使用Flink读取"avro“序列化数据的Kafka流，如下所示： tableEnv.connect(new Kafka()
.version("0.11")
.topic(source.getTopic())
.properties(source.getProperties())
.startFromLatest())
.withSchema(Schemafy.getSchemaFromJs
浏览 359提问于2019-06-05得票数 2
我的客户类已经使用maven-avro plugin.When创建了，我尝试运行这个程序，我得到的是Exception in thread "main" java.lang.IllegalStateException: Expecting type to be a PojoTypeInfo错误。
[main] INFO org.apache.flink.api.java.typeutils.TypeExtractor - class com.example.Customer does not contain a setter for field first_name
[main
浏览 26提问于2021-08-18得票数 1
我已经编写了一个使用者来使用模式注册表读取Avro的通用记录。
FlinkKafkaConsumer010 kafkaConsumer010 = new FlinkKafkaConsumer010(KAFKA_TOPICS,
new KafkaGenericAvroDeserializationSchema(schemaRegistryUrl),
properties);
反序列化类如下所示：
public class KafkaGenericAvroDeserializationSchema implements KeyedDe
浏览 0提问于2020-01-23得票数 1
我是flink和Kafka的新手。我正在尝试使用Confluent Schema注册表对avro数据进行反序列化。我已经在ec2机器上安装了flink和Kafka。此外，在运行代码之前已经创建了"test“主题。 代码路径：https://gist.github.com/mandar2174/5dc13350b296abf127b92d0697c320f2 作为实现的一部分，代码执行以下操作： 1) Create a flink DataStream object using a list of user element. (User class is avro generated
浏览 43提问于2019-04-22得票数 1
1回答
在我的Flink反序列化程序中，我从getProducedType方法中得到以下错误：
com.esotericsoftware.kryo.KryoException: java.lang.NullPointerException
Serialization trace:
values (org.apache.avro.generic.GenericData$Record)
at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:125)
at com.esotericsoftware
浏览 31提问于2018-01-03得票数 3
回答已采纳
我有一个Flink作业，它在本地运行得很好，但当我尝试在集群上运行flink作业时失败。当尝试通过'connector‘= ' Kafka’从kafka加载数据时发生错误。我使用Flink-Table API和confluent-avro格式从Kafka读取数据。 所以基本上我创建了一个从kafka主题中读取数据的表： val inputTableSQL =
s"""CREATE TABLE input_table (
| -- key of the topic
| key BYTES NOT
浏览 299提问于2021-10-22得票数 0
回答已采纳
我得到了阿夫罗的回应，从卡夫卡的话题汇合，我面临的问题，当我想要得到的回应。不理解语法，我应该如何定义阿夫罗反序列化器和使用在我的卡夫卡源，同时阅读。
分享我目前正在做的方法。
在Confluent中，我有一个名为employee的主题，它每10秒生成一次消息，每条消息都由Confluent中的avro模式注册中心进行服务器化。
我试图在scala程序中读取这些消息，我能够在代码中打印序列化的消息，但无法反序列化消息。
import org.apache.flink.streaming.api.scala._
import org.apache.flink.api.common.eventti
浏览 12提问于2022-05-06得票数 0
我真的很难让Flink与运行中的Kafka实例进行正确的通信，使用来自communicate的Avro模式(对于，键和值)。
经过一段时间的思考和重组我的计划，我能够推动我的实现到目前为止：
生产方法
public static FlinkKafkaProducer<Tuple2<GenericRecord,GenericRecord>> kafkaAvroGenericProducer() {
final Properties properties = new Properties();
properties.put(Pro
浏览 11提问于2020-01-30得票数 0
回答已采纳
2回答
我正在尝试使用AVRO 1.0.3从Kafka主题读取Flink数据。
我只知道这个特别的Kafka主题是有AVRO编码的消息，我有AVRO模式文件。
我的Flink代码：
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
Properties properties = new Properties();
浏览 3提问于2016-08-02得票数 3
回答已采纳
我正在编写一个Apache流应用程序，它反序列化数据(Avro格式)，读取一个Kafka总线(关于的更多细节)。数据正在反序列化为Scala案例类。当我运行这个程序时，我得到了一个异常，它收到了Kafka的第一条消息。
Exception in thread "main" org.apache.flink.runtime.client.JobExecutionException: java.lang.RuntimeException: java.lang.NoSuchMethodException: org.myorg.quickstart.DeviceData.<ini
浏览 1提问于2018-07-02得票数 5
回答已采纳
据我所知，使用Flink的AVRO反序列化，您可以创建一个Avro对象流，这很好，但似乎存在一个问题，Flink的kafka使用者只创建单个对象的流：FlinkKafkaConsumerBase<T>而不是默认的Kafka及其KafkaConsumer。
在我的例子中，键和值都是独立的符合AVRO模式的对象，合并它们的模式可能是一场噩梦.
另外，使用Flink，我似乎无法检索ConsumerRecord信息？
浏览 3提问于2018-11-15得票数 0
回答已采纳
1回答
我正在Scala中构建一个Apache应用程序，它从Kafka总线读取流数据，然后对其执行汇总操作。Kafka的数据是Avro格式的，需要一个特殊的反序列化类。我找到了这个scala类AvroDeserializationScehema ()：
package org.myorg.quickstart
import org.apache.avro.io.BinaryDecoder
import org.apache.avro.io.DatumReader
import org.apache.avro.io.DecoderFactory
import org.apache.avro.reflec
浏览 1提问于2018-07-01得票数 0
回答已采纳
我已经编写了一个消费者，它读取卡夫卡主题，并使用StreamSink以拼图格式写入数据。但是我得到了以下错误
java.lang.Exception: org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException: Could not forward element to next operator
at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.checkThrowS
浏览 6提问于2020-02-29得票数 0
我有一个带有以下Avro IDL的Kafka主题，并注册到Schema Registry。 @namespace("my.type.avro")
protocol MyProtocol {
enum MyEnumType {
Type1, Type2
}
record MyEntry {
MyEnumType myEntryType = "Type1";
}
record MyRecord {
MyEntry entry;
浏览 190提问于2021-10-12得票数 1
回答已采纳
我想对Flink SQL表执行一个查询，该查询由安全的kafka集群的kafka主题支持。我能够以编程方式执行查询，但无法通过Flink SQL客户端执行。我不确定如何通过Flink传递JAAS (java.security.auth.login.config)和其他系统属性。
Flink SQL查询以编程方式
private static void simpleExec_auth() {
// Create the execution environment.
final EnvironmentSettings settings = Environment
浏览 1提问于2021-02-23得票数 1
我正在使用集群、Schema服务，并试图在Flink作业中处理Debezium消息。作业被配置为使用Table & SQL连接器和Confluent格式。但是，作业无法连接到架构注册表并引发401错误。
表连接器配置
tEnv.executeSql("CREATE TABLE flink_test_1 (\n" +
" ORDER_ID STRING,\n" +
" ORDER_TYPE STRING,\n" +
" USER_ID STRING,\n" +
浏览 3提问于2021-02-08得票数 3
1回答
我使用flink流从Kafka源主题中读取事件，在去复制之后，编写来将avro主题中的kafka主题分离开来。流
卡夫卡主题(json格式) -> flink流(去复制) -> scala类对象-> Kafka主题(Avro格式)
val sink = sinkProvider.getKafkaSink(brokerURL, targetTopic,kafkaTransactionMaxTimeoutMs, kafkaTransactionTimeoutMs)
messageStream
.map {
record =>
浏览 6提问于2022-11-20得票数 0
我正在尝试读取Kafka主题中的记录，该主题是由kafka connect jdbc源连接器生成的，以下是连接器配置： "name": "customers",
"config": {
"poll.interval.ms": "3000",
"table.poll.interval.ms": "10000",
"errors.log.enable": "false",
"errors.log.include.m
浏览 670提问于2021-02-01得票数 0
回答已采纳
1回答
我们正在使用一个集成平台来部署Kafka。我们使用模式注册表来存储模式。是否可以将模式注册中心与flink集成？如何从汇流平台读取AVRO格式的数据？
浏览 8提问于2022-06-17得票数 0
回答已采纳
由于flink问题：，我正在尝试手动创建一个使用者组，这样flink作业就可以成功地运行，但是使用下面的代码，我无法创建一个，是不是遗漏了什么。
val properties = new Properties()
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, broker)
properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId)
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, resetMode)
proper
浏览 4提问于2022-01-19得票数 1
我有一个容器化的flink集群，具有独立的JobManager和2个任务管理器。当我提交一个包含我需要的连接器和代码的胖jar时，一切都会正常运行，但是一个只有我的代码的jar将失败--尽管我已经将连接器jar添加到/opt/flink/lib，容器进程显示它们将被添加到类路径中。当我试图从IDE进行远程连接时，也会发生同样的失败；不用说，每次都要打包一个胖jar，这会带来糟糕的开发体验。
我需要做什么才能让flink理解我放置在/opt/flink/lib中的额外jars应该与所有新作业一起使用？
类加载程序异常：
org.apache.flink.streaming.runtime.ta
浏览 0提问于2019-04-12得票数 1
回答已采纳
我正在写一个流应用程序(Flink，Kafka，S3)，它把数据放在Kafka上的JSON格式，我需要写数据到S3的Parquet格式。
我搜索了很多地方，读了很多关于stackoverflow的问题，但我没有得到一个明确的答案，那就是如果我想把拼图写到s3中，是不是必须在flink中读取AVRO格式的数据？
我需要从JSON -> AVRO ->拼花？
或者使用kafka、flink、S3也可以使用JSON ->拼图
浏览 47提问于2020-02-25得票数 0
作为我的flink应用程序的运行时，我通过AWS ()使用托管flink
我在S3中添加了功能(接收器)，用于将处理过的事件从动态队列中以拼花格式写入。在本地，一切都适合我，但是当我试图在云中运行应用程序时，我得到了以下例外：
"throwableInformation": [
"com.esotericsoftware.kryo.KryoException: Error constructing instance of class: org.apache.avro.Schema$LockableArrayList",
&
浏览 2提问于2020-06-03得票数 1
回答已采纳
作为PyFlink pipeline.classpath配置的一部分传递的类加载类与将它们放入$FLINK_HOME\lib目录之间有什么区别？
当我想使用flink-sql-connector-kafka-*.jar时，只要使用pipeline.classpath传递它就可以了，但是当我想使用一些具有一些外部依赖项的东西时，比如需要avro-*.jar jars的flink-avro-*.jar。它似乎加载了flink-avro-*.jar，但它似乎无法加载avro-*.jar并抛出：
java.lang.NoClassDefFoundError: Could not initialize
浏览 15提问于2022-01-24得票数 0
回答已采纳
最近，我尝试从Flink 1.3.2升级到1.4.0，我遇到了一些问题，无法再导入org.apache.hadoop.fs.{FileSystem, Path}。这一问题发生在两个地方：
ParquetWriter：
import org.apache.avro.Schema
import org.apache.avro.generic.GenericRecord
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.flink.streaming.connectors.fs.Writer
import org.apa
浏览 9提问于2017-12-27得票数 2
回答已采纳
场景: Apache Flink、Kafka、Protocol buffers data consumer。 数据来源为协议缓冲区格式的Kafka主题(多个主题: topic#1、topic#3、topic#3)。使用者是Apache Flink使用者。每个主题都有一个独特的协议定义。 List<String> topicList = Arrays.asList("topic#1,topic#2,topic#3".split(","));
inputStream = env.addSource(new FlinkKafkaConsumer[Prot
浏览 13提问于2021-04-03得票数 0
我正在尝试写案例类序列化到AVRO到HDFS使用RollingSink在Flink。为了让avro文件可以被HDFS反序列化，我使用了包装FSDataOutputStream的DataFileWriter。当我试图在DataFileWriter和FSDataOutputStream之间同步以便关闭HDFS上的数据文件时，抛出异常，实际上我每隔一个文件就会得到数据。有没有办法在Flink writer实现中同步fs流和Avro编写器？
我尝试过使用DataFileWriter close()、flush()、sync()、fsync()，但都失败了。Sync方法似乎执行得最好。我也尝试过在wri
浏览 1提问于2016-03-24得票数 0
我有一个flink应用程序，它从kafka中读取并将其吸收到kafka。
当我从Intellij运行应用程序时，没有问题，但是当我将ClassCastException提交给flink集群时，它会给出shadowJar。我能在找出我在这里做错了什么方面得到一些帮助吗？
异常跟踪：
Caused by: java.lang.ClassCastException: cannot assign instance of org.apache.kafka.clients.consumer.OffsetResetStrategy to field org.apache.flink.connector.ka
浏览 61提问于2022-05-16得票数 4
我使用io.confluent.kafka.streams.serdes.avro.GenericAvroDeserializer反序列化卡夫卡Message<Bytes>，它的卡夫卡"value.deserializer“配置为org.apache.kafka.common.serialization.BytesDeserializer，但GenericAvroDeserializer没有反序列化以下类型的avro有效负载
{
"name": "totalAmount",
"type": [
浏览 6提问于2021-12-06得票数 1
2回答
我得到了以下SBT文件：
.
-- root
-- plugins.sbt
-- build.sbt
其中plugins.sbt包含以下内容：
addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.12.0")
和包含以下内容的build.sbt：
import sbt.Keys._
resolvers in ThisBuild ++= Seq("Apache Development Snapshot Repository" at "https://reposito
浏览 1提问于2016-08-09得票数 1
1回答
我试图更改Flink运行程序代码，让它读取来自几个Kafka主题的数据，并相应地将其写入不同的HDFS文件夹，而无需加入。我有很多Java和Scala泛型方法，以及主进程方法和反射内部的泛型对象初始化。它适用于一个Avro模式，但是当我试图添加未知数量的Avro模式时，泛型和反射结构会出现问题。
如何解决这个问题？设计模式对我有什么帮助？
模型(Avro模式)在Java类中。
public enum Types implements MessageType {
RECORD_1("record1", "01", Record1.getClassS
浏览 2提问于2020-02-04得票数 0
假设我通过kafka streams从Confluent创建了主题，其中包含使用io.confluent.kafka.streams.serdes.avro.SpecificAvroSerializer在avro中序列化的消息
然后在Hive中创建一个外部kafka表
CREATE EXTERNAL TABLE k_table
(`id` string , `sequence` int)
STORED BY 'org.apache.hadoop.hive.kafka.KafkaStorageHandler'
TBLPROPERTIES
(
"kafka.to
浏览 2提问于2019-02-06得票数 0
我已经实现了这里建议的解决方案：，所以我的代码如下所示：
Schema schema = Schema.parse("{\"type\":\"record\",\"name\":\"Envelope\" ... etc");
DataStreamSource<GenericRecord> stream = env.addSource(new FlinkKafkaConsumer<>(topic, AvroDeserializationSchema.forGeneric(schema), ge
浏览 3提问于2021-08-18得票数 1
回答已采纳
我正在寻找一个debezium连接器来将CDC记录流到kafka，其中键为字符串(而不是avro表示键)，值为avro记录。默认情况下，它将密钥作为avro记录。有什么建议吗？
浏览 0提问于2018-06-12得票数 2
回答已采纳
我使用Flink的表API创建了一个表。
CREATE TABLE recommendations (
...
) WITH (
'connector' = 'kafka',
'topic' = 'my_kafka_topic',
'properties.bootstrap.servers' = 'localhost:9092',
'properties.group.id' = 'testGroup',
'properties.secu
浏览 0提问于2021-10-12得票数 0
创建Kafka JdbcSinkConnector时出错(我的任务是将数据从Kafka主题传输到Postgres表)：
导致: org.apache.kafka.common.errors.SerializationException:错误反序列化id -1的Avro消息
身份证是什么意思？
连接器设置：
{
"name": "MVM Test",
"connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
"key.convert
浏览 3提问于2020-01-21得票数 0
回答已采纳
2回答
我正试图使用FlinkKafkaProducer10从Flink中将kafka与模式注册表汇合在一起。产生下面的错误。我创建了自定义模式序列化器(请参阅ConfluentAvroSerializationSchema类)。代码编译但会产生运行时错误。生成错误的完整代码示例在上。所有资源都是模拟的，kafka集群和模式注册表。代码实际上是测试用例。
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import org.apache.flink.streaming.util.serialization.Serialization
浏览 4提问于2017-11-19得票数 4
1回答
我是Scala和Apache的初学者，但到目前为止，一切都对我有利。我正在尝试使用从我的Flink应用程序序列化到AVRO的Kafka事件。我阅读了文档()并搜索了许多小时，但我仍然在同一页上。我有一个case类case class URLResponse(status: int, domain: String, url: String, queue: String, html: String)和一个模式val schema: Schema = new Schema.Parser().parse("{\"type\": \"record\", \"
浏览 3提问于2021-01-19得票数 0
我正在尝试创建一个kafka avro序列化主题的flink消费者。我有kafka主题流式avro序列化数据。我可以通过avroconsoleconsumer看到它。
Flink 1.6.0增加了一个AvroDeserializationSchema，但我找不到一个完整的用法示例。是的，有几个似乎在1.6.0之前生成了一个反序列化类，这个类补充道。
我有一个通过avro-tools生成的avro类。
现在，我一直在尝试遵循现有的例子，但它们之间的差异足够大，以至于我无法让事情继续下去。(我不经常用Java编程)
大多数都使用下面的某种形式
Myclass mc = new MyClass();
浏览 15提问于2018-08-20得票数 0
1回答
我将avro序列化数据发布到kafka主题，然后尝试通过SQL接口从主题创建Flink表。我能够创建主题，但在执行SQL SELECT语句后无法查看主题数据。但是，我能够使用简单的kafka用户反序列化和打印已发布的数据。在SQL上获取此错误：
Flink SQL> SELECT * FROM test_flink2;
[ERROR] Could not execute SQL statement. Reason:
java.lang.ArrayIndexOutOfBoundsException: Index -3 out of bounds for length 2
表创建
Flink
浏览 4提问于2021-02-05得票数 1
回答已采纳
我使用Kafka作为Flink job的数据源。当我将作业部署到flink集群作业管理器时，我收到一个错误ClassNotFoundException：Caused by: java.lang.ClassNotFoundException: org.apache.flink.connector.kafka.source.KafkaSource 下面是我的pom.xml依赖项 <dependencies>
<dependency>
<groupId>org.springframework.boot</groupI
浏览 489提问于2021-08-28得票数 0
回答已采纳
1回答
我试图使用avro模式将数据从logstash发送到kafka。
我的logstash输出看起来如下：
kafka{
codec => avro {
schema_uri => "/tmp/avro/hadoop.avsc"
}
topic_id => "hadoop_log_processed"
}
我的模式文件看起来像：
{"type": "record",
"name": "hadoop_schema",
"fields": [
浏览 5提问于2020-04-22得票数 1
回答已采纳
我在网上到处查找，发现这是一个很常见的错误，但没有任何解决方案对我有帮助。
我在读卡夫卡的一个主题。到目前为止，我还没有遇到这样的问题，但是现在，当我在aws环境中运行flink集群而不是在我的IDE (intellij)上运行时，我得到了这个错误的。
NoClassDefFoundError: org/xerial/snappy/Snappy
at org.apache.avro.file.SnappyCodec.decompress(SnappyCodec.java:58)
at org.apache.avro.file.DataFileStream$DataBlock.decompres
浏览 3提问于2017-06-20得票数 2
回答已采纳
我想在apache中序列化和反序列化avro消息。我也很乐意在Scala中实现。
我已经在下面附上了我的源代码。我希望源代码中的注释足够清楚。
package com.example;
// import io.confluent.kafka.serializers.KafkaAvroSerializer;
// import org.apache.flink.api.common.serialization.SimpleStringSchema;
import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import io.
浏览 5提问于2022-04-19得票数 0
因此，我有一个卡夫卡主题，其中包含了不同图式的avro记录。我想使用flink中的Kafka主题，并创建avro泛型记录的数据。
现在，我想使用从datastream提取的模式在hudi中编写数据。但是由于hudi管道/编写器在一开始就使用预定义的avro模式进行配置，所以我不能这样做。
可能的解决方案是基于一个键创建一个密钥流，他们识别一种模式类型，然后从其中提取模式，然后基于它创建一个动态hudi管道。
我不确定最后那部分是否可能。
A->B->C
其中A是具有不同模式的泛型avro记录。B是基于不同模式的分区流。C在数据流B中使用该模式创建配置，并将其传递给hudi管道编写器
浏览 9提问于2022-10-20得票数 0
1回答
嗨，我正在尝试运行一个flink scala应用程序，它从kafka读取，应用一些查找转换，然后写到kafka。
Flink版本1.12.1
我在当地测试过，效果很好。但是，当我尝试使用原生kubernetes集成在集群上运行它时，我会看到下面这样的奇怪错误。
集群看起来也很好，因为我试图在集群上运行一个wordcount应用程序，它运行得很好。
异常不清楚，堆栈跟踪也显示了任务管理器堆栈跟踪，因此不知道应用程序中的问题可能在哪里。这可能是序列化问题吗？是否有一种方法来调试这些问题，并在应用程序代码中找到有问题的实际点？
org.apache.flink.streaming.runtime.t
浏览 2提问于2021-09-01得票数 0
回答已采纳
我想就以下问题寻求一些指导。我正在学习如何在没有模式注册的情况下使用nodejs执行Avro数据的序列化，将其发布到Kafka集群，然后在Kafka Streams (Java)中检索它。
在javascript方面，我尝试使用kafka节点和avsc进行序列化。在Kafka流中，我决定实现一个自定义Serde，因为据我所知，由Streams API提供的Avro Serdes被设计为直接从Schema注册表获取模式。
下面是一个简单生产者的javascript代码片段：
const avro = require('avsc');
const messageKey =
浏览 3提问于2020-03-06得票数 0
回答已采纳
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM
腾讯云开发者
