我的客户类已经使用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
我已经从postgresql DB.then加载了一个规则表作为Flink表,读取kafka msg,并根据这些规则对msg进行分类。代码如下所示 val senv = StreamExecutionEnvironment.getExecutionEnvironment
senv.enableCheckpointing(5000)
val stenv=StreamTableEnvironment.create(senv)
val streamsource=senv.createInput(inputFormat)
stenv.registerDataS
我使用dataSet API,我有两种案例类。
case class Geo(country:Int, province:Int, city:Int, county:Int)
case class AntiFraudLog(
eventType: Int,
valid: Boolean
)
case class AntiFraudSession(fraudLogs: Seq[AntiFraudLog])
然后我生成了一个键/值对,它的值是一个case类。
val dataKeyValue: DataSet[(Long, AntiFraudLog)]
并尝试使
我正在尝试用case类阅读avro kafka的Scala主题:
val senv = StreamExecutionEnvironment.getExecutionEnvironment
val kafkaSourceTopic = "device_logs"
val kafkaBrokers = "localhost:9092"
val properties = new Properties()
properties.setProperty("bootstrap.servers", kafkaBrokers)
properties.set
我尝试将我的Flink应用程序部署到AWS中。此应用程序使用Apache进行反序列化/序列化传入消息。我的应用程序在本地机器上运行良好,但是当我将它部署到AWS时,我有异常(在CloudWatch日志中):Caused by: java.io.InvalidClassException: org.apache.avro.specific.SpecificRecordBase; local class incompatible: stream classdesc serialVersionUID = 4445917349737100331, local class serialVersionUI
有时会引发此错误,并向下查看任务管理器。我使用了org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode WARN org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer - Falling back to default Kryo serializer because Chill serializer couldn't be found.
java.lang.reflect.InvocationTargetExc
我是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
FLINK流:,我有来自kafkaconsumer的DataStreamString,这是JSON
stream = env
.addSource(new FlinkKafkaConsumer[String]("topic", new SimpleStringSchema(), properties))
我必须使用StreamingFileSink (需要DataStreamGenericRecord )来接收这个流。
val schema: Schema = ...
val input: DataStream[GenericRecord] = ...
val sin
我最近开始“玩”Apache Flink。我已经组装了一个小应用程序来开始测试框架等等。当我尝试序列化一个普通的POJO类时,我现在遇到了一个问题: @Getter
@ToString
@EqualsAndHashCode
@NoArgsConstructor
@AllArgsConstructor
public final class Species {
private String name;
private List<String> abilities;
} 不知何故,我可以从堆栈跟踪看出List类型不能序列化,但根据Flink的文档,情况不应该是这样的。这是堆栈
我将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
我们有一个用Scala编写的Flink作业,使用case类(由avrohugger从avsc文件生成)来表示我们的状态。我们希望使用Avro来序列化我们的状态,这样当我们更新我们的模型时,状态迁移就会起作用。我们从Flink 1.7开始就知道Avro序列化是支持OOTB的。我们将flink-avro模块添加到类路径中,但从保存的快照恢复时,我们注意到它仍在尝试使用Kryo序列化。相关代码片段 case class Foo(id: String, timestamp: java.time.Instant)
val env = StreamExecutionEnvironment.getExe
我正在尝试制作一些scala函数,这些函数可以帮助flink、map和filter操作将错误重定向到死信队列。
然而,我正在与scala的类型擦除做斗争,这使我无法使它们成为通用的。下面mapWithDeadLetterQueue的实现不编译。
sealed trait ProcessingResult[T]
case class ProcessingSuccess[T,U](result: U) extends ProcessingResult[T]
case class ProcessingError[T: TypeInformation](errorMessage: String, ex
我得到了几个Avro文件,它们是从java类A的对象生成的。使用spark-shell (Spark 1.4.0),我可以使用spark-avro (2.0.1版)将这些文件读取到数据帧中。
val df = sqlContext.read.avro("file.avro")
只要spark shell不知道类A,就可以正常工作。如果我向shell添加一个jar,并发出相同的命令,我会得到以下异常:
A cannot be cast to org.apache.avro.generic.IndexedRecord
at org.apache.avro.generic.G
我真的很难让Flink与运行中的Kafka实例进行正确的通信,使用来自communicate的Avro模式(对于,键和值)。
经过一段时间的思考和重组我的计划,我能够推动我的实现到目前为止:
生产方法
public static FlinkKafkaProducer<Tuple2<GenericRecord,GenericRecord>> kafkaAvroGenericProducer() {
final Properties properties = new Properties();
properties.put(Pro
我目前正在开发一个从Hadoop集群读取Avro文件的Flink 1.4应用程序。但是,在我的IDE上以本地模式运行它工作得很好。但是,当我将其提交到Jobmanager Flink时,它总是失败,并显示以下消息:
java.io.IOException: Error opening the Input Split hdfs://namenode/topics/CaseLocations/partition=0/CaseLocations+0+0000155791+0000255790.avro [0,16549587]: Could not find a file system implem
我在Scala (2.12.8) Apache (1.9.1)应用程序中使用case类。当我在Caused by: java.lang.NoSuchMethodError: scala.Product.$init$(Lscala/Product;)V.下面运行代码时,我会得到以下异常
注意:我按照建议( )使用了默认构造函数,但在我的示例中不起作用。
这是完整的代码
package com.zignallabs
import org.apache.flink.api.scala._
/**
// Implements the program that reads from a Element
我正在使用Linux中的Docker运行一个带有单个节点的Flink独立集群。我已经在Flink 1.10.0和JDK8的生产环境中运行了一段时间的前一个版本,我能够让S3在那里正常运行。现在,我正在尝试更新到一个较新的版本,使用本地S3实现在我的开发机器上运行Docker。不管我怎么尝试,这个错误总是弹出: org.apache.flink.core.fs.UnsupportedFileSystemSchemeException: Could not find a file system implementation for scheme 's3'. 看起来S3方案没有被映
我正在尝试写一些输出到S3使用电子病历与Flink。我使用的是Scala 2.11.7、Flink 1.3.2和EMR 5.11。但是,我得到了以下错误:
java.lang.NoSuchMethodError: org.apache.hadoop.conf.Configuration.addResource(Lorg/apache/hadoop/conf/Configuration;)V
at com.amazon.ws.emr.hadoop.fs.EmrFileSystem.initialize(EmrFileSystem.java:93)
at org.apache.fl