使用Java通过Apache Flink减少Pojo字段
Apache Flink是一个开源的流处理和批处理框架，它提供了高效、可扩展和容错的数据处理能力。通过使用Java编程语言，结合Apache Flink，可以减少Pojo字段。
Pojo（Plain Old Java Object）是指普通的Java对象，它是一个简单的Java类，不继承或实现任何特殊的接口。在数据处理中，我们经常需要对数据进行转换、过滤、聚合等操作，而这些操作通常需要对数据的字段进行处理。使用Apache Flink可以方便地对Pojo对象的字段进行操作和转换。
具体来说，使用Java通过Apache Flink减少Pojo字段可以通过以下步骤实现：
- 定义Pojo类：首先，需要定义一个包含需要处理的字段的Pojo类。例如，假设我们有一个Person类，包含name、age和gender字段。
- 创建数据流：使用Apache Flink的DataStream API，可以从各种数据源（如文件、消息队列、数据库等）创建数据流。可以使用Flink提供的各种数据源连接器，如Kafka Connector、JDBC Connector等。
- 转换操作：通过使用Apache Flink提供的转换操作，可以对数据流进行各种操作，包括过滤、映射、聚合等。对于减少Pojo字段，可以使用map()或flatMap()操作来选择需要的字段，或者对字段进行转换。
- 输出结果：最后，可以将处理后的数据流输出到目标位置，如文件、数据库、消息队列等。可以使用Flink提供的各种Sink连接器，如File Sink、JDBC Sink等。
Apache Flink的优势在于其高性能、可扩展性和容错性。它采用了流式计算模型，能够处理实时数据和批量数据，并且具有低延迟和高吞吐量的特点。此外，Flink提供了丰富的API和库，支持复杂的数据处理操作，如窗口计算、状态管理、事件时间处理等。
对于使用Java通过Apache Flink减少Pojo字段的应用场景，可以包括实时数据处理、流式ETL、实时分析等。例如，在电商领域，可以使用Flink来处理实时的用户行为数据，提取关键字段进行实时分析和推荐。
腾讯云提供了云原生的数据计算服务Tencent Cloud TKE Flink，可以方便地在云上部署和管理Apache Flink集群。您可以通过以下链接了解更多关于Tencent Cloud TKE Flink的信息：Tencent Cloud TKE Flink
总结：通过Java编程语言结合Apache Flink，可以方便地减少Pojo字段。Apache Flink是一个高性能、可扩展和容错的流处理和批处理框架，适用于实时数据处理、流式ETL、实时分析等场景。腾讯云提供了云原生的数据计算服务Tencent Cloud TKE Flink，方便用户在云上部署和管理Flink集群。
相关·内容
我们的一些POJO包含来自java.time接口(LocalDate，LocalDateTime)的字段。当我们的管道处理它们时，我们可以在日志中看到以下信息：
org.apache.flink.api.java.typeutils.TypeExtractor - Class class java.time.LocalDate cannot be used as a POJO type because not all fields are valid POJO fields, and must be processed as GenericType. Please read the Flink
浏览 240提问于2021-02-17得票数 1
回答已采纳
在集成测试中测试flink mini集群上的流时，我遇到了一个问题。该流映射一个生成的Avro Java类( SpecificRecord )。 流作业是用Scala编写的。 flink运行时正在崩溃，因为它无法实例化org.apache.flink.formats.avro.utils.AvroKryoSerializerUtils 下面是堆栈跟踪： stack: java.lang.ClassCastException: class org.apache.flink.formats.avro.utils.AvroKryoSerializerUtils
java.lang.RuntimeE
浏览 38提问于2020-09-25得票数 0
在flink项目中，我使用case类单击。
case class click( date: LocalDateTime, stbId:String, channelId :Int)
这个类填充了DataSets，并且在日期为java 8 java.time.LocalDateTime的情况下工作得很好。在java7环境中切换到org.joda (版本2.9)之后，对DataSets中的click对象的调用没有像以前那样执行。访问单击对象的日期字段的某些函数会引发NullPointerExceptions。这些函数的例子是getHourOfDay、toString等。我能够确保click类的da
浏览 11提问于2015-11-11得票数 4
回答已采纳
我正在尝试做一个POC的Flink状态模式演变。我使用的是Flink 1.15.0和Java 11。我尝试创建3个数据类--每个序列化类型一个：
io.peleg.kryo.User -使用java.time.Instant类，我知道Flink中的POJO序列化不支持这个类。
io.peleg.pojo.User -只使用经典的包装原语- Integer、Long、String。getters、setter和constructors是使用Lombok生成的。
io.peleg.avro.User -使用Avro插件从Avro生成。
对于每个类，我编写了一个流作业，它使用一个
浏览 66提问于2022-07-02得票数 2
我正在用Kotlin编写Flink应用程序，数据类(以及其他Kotlin类)没有标识为有效的POJO类型。
Flink文档指出，如果满足以下条件，数据类型将被识别为POJO类型(并允许“按名称”字段引用)：
这门课是公开的和独立的。
这个类有一个公共的无参数构造函数。
类中的所有非静态、非瞬态字段要么是公共的(也不是最终的)，要么具有遵循Java命名约定的公共getter和setter方法。
在实现Kotlin数据类时，我收到以下信息，该类应满足上述条件，被确认为POJO：
[main] INFO org.apache.flink.api.java.typeutils.Ty
浏览 15提问于2019-01-18得票数 4
回答已采纳
不是一个错误，但我确实看到了这一行，根据消息可能会影响性能： 2019-01-02 14:44:44,879 INFO org.apache.flink.api.java.typeutils.TypeExtractor
- class org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition
does not contain a setter for field topic
2019-01-02 14:44:44,879 INFO org.apache.flink.api
浏览 123提问于2019-01-02得票数 0
我使用的是带有32 32执行程序的flink mysql连接器，有32个插槽的16 16vCPU。如果我运行一个具有并行性32 (作业并行性224)的作业，它使用10个MySQL表执行时态查找联接，那么在2-3次成功运行之后，它就会开始失败。
org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy
at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandle
浏览 17提问于2022-01-17得票数 0
2回答
我试图在Flink中的KeyedStream上执行地图操作：
stream.map(new JsonToMessageObjectMapper())
.keyBy("keyfield")
.map(new MessageProcessorStateful())
JsonToObjectMapper操作符的输出是MessageObject类的POJO，它有一个字符串字段'keyfield‘。然后在这个字段上输入流。
MessageProcessorStateful是一个像这样的RichMapF
浏览 0提问于2018-02-01得票数 4
回答已采纳
我的Flink管道目前使用一个Pojo，其中包含一些列表和映射( Strings)，按照
public class MyPojo {
private List<String> myList = new ArrayList<>();
private OtherPojo otherPojo = new OtherPojo();
// getters + setters...
}
public class OtherPojo {
private Map<String, String> myMap = new HashMap<
浏览 16提问于2020-01-18得票数 3
回答已采纳
我有scala类，它有两个字段，即、vals、，但是flink说它没有setter。因此，任务是不可序列化的。
我试过使用var来设置setter，但是后面写的是重复的setter。vals是公开的，那么它为什么要邀请策划人。Flink版本1.1.0
class Impression(val map: Map[String, String],val keySet:Set[String])
我的代码：
val preAggregate = stream
.filter(impression => {
true
})
.map(impression => {
val
浏览 7提问于2016-08-08得票数 3
3回答
我有一个类可以满足作为POJO处理的需求，这是流作业中的主要传输类(它只包含原语和Map<String, String>)。我添加了一个新的String字段和相应的getter和setter，但是如果我停止使用带有保存点的前一个类的作业，并尝试使用它重新启动新类，则会得到一个异常：
java.lang.Exception: Exception while creating StreamOperatorStateContext.
at org.apache.flink.streaming.api.operators.StreamTaskStateInitialize
浏览 31提问于2020-11-05得票数 0
回答已采纳
我试图从cassandra表中提取数据作为数据集，但我遇到了两个问题。
首先，cassandraInputFormat只返回一个元组，我更希望没有tuple12，只需要使用pojo来定义它期望返回的内容。因此，我不知道这是否只是我必须接受的东西，是否有一种方法可以使用pojo代替cassandraConnector ()，或者使用cassandraInputFormat不是获取数据的最佳方式。
另一个问题是，即使我从cassandraInputFormat中提取的数据(不管是元组还是非元组)，我也不知道如何将它设置为数据源。对于文件、csv和HDFS，有很多方法()，但没有一个方法显式地用于c
浏览 4提问于2017-05-11得票数 0
回答已采纳
在我的Flink应用程序中，我使用java.time.Instant来表示UTC时间戳。应用程序运行正常，但我最近在Flink日志中注意到了这条消息：
“类java.time.Instant不能用作POJO类型，因为并非所有字段都是有效的POJO字段，必须作为GenericType处理。有关对性能的影响的详细信息，请阅读\”数据类型和序列化\“上的Flink文档。”
当我去阅读文档时，没有太多关于使用像Instant这样的东西对性能的影响的讨论。我一般的理解是，必须使用Kryo而不是Flink内置的序列化程序。我目前使用的是Flink 1.6，我发现Flink 1.7及更高版本似乎有一个Ins
浏览 93提问于2020-02-05得票数 0
回答已采纳
相关问题：https://issues.apache.org/jira/browse/FLINK-14380 在使用类型提取器Flink判断一个类是否符合POJO时，Flink会检查该类是否实现了setter和getter方法。对于setter方法，Flink断言返回类型为Void。如果经常使用case类，case类setter的返回类型是objects类的副本，这是一个问题。考虑以下case类： case class SomeClass(x: Int) {
x_=(newX: Int): SomeClass = { this.copy(x = newX) }
} 这个类将被标识为
浏览 37提问于2019-10-12得票数 2
使用Flink 1.12。 我的协议是 syntax = "proto3";
package flink.protobuf;
message TimestampedMessage {
int64 timeMs = 1;
string message = 2;
} 并尝试像这样使用它 final var env = StreamExecutionEnvironment.createLocalEnvironment();
env.getConfig().registerTypeWithKryoSerializer(Timestamp
浏览 63提问于2021-08-12得票数 0
请帮帮我，我有一个集群Apache (2个作业经理，3个任务管理器)，但我不知道在Flink -conf.yml. for中为该参数设置哪些值：
jobmanager.heap.size taskmanager.heap.size taskmanager.numberOfTaskSlots parallelism.default
作业管理器机器有:8 8CPU，32 Job
任务管理器机器有:8 8CPU，32 RAM
我计划在这个集群15..20 Apache乔布斯上运行。由于私有策略，我无法在这里编写java代码，因此我将尝试用语言表达。
1)我从Apache№1读取数据(这是
浏览 3提问于2020-06-30得票数 0
回答已采纳
我对Flink的表-api (1.13+)有一些问题.我有一个包含几个字段的POJO，其中之一是：
List<String> my_list;
我为这个字段使用以下声明来创建我的表：
"CREATE TABLE my_table (
...
my_list ARRAY<STRING>,
...
)"
之后，当我试图使用DataStream方法将表转换为toRetractStreamMY_POJO时，我得到了以下错误：
未注册表的查询结果和接收器的线程“主”org.apache.flink.table.api.ValidationEx
浏览 2提问于2022-01-19得票数 5
当使用非POJO的java对象的ValueState时，如HyperLogLogPlus，如何处理序列化问题？
我曾尝试定义HyperLogLogPlusSerializer扩展TypeSerializer，代码如下：
public class HyperLogLogPlusSerializer extends TypeSerializer<HyperLogLogPlus> {
private static Logger logger = LoggerFactory.getLogger(HyperLogLogPlusSerializer.class);
priv
浏览 0提问于2019-05-29得票数 2
1回答
我在flink中使用广播状态模式，在这里我试图连接两个流，一个流是规则的控制流，另一个流是整数流(用于虚拟播放目的)。
我有一个Rule类
public class Rule {
String id;
int val;
RuleType ruleType;
//Newly added field
//int val2 = 0;
public Rule() {}
public Rule(String id, int val, RuleType ruleType) {
this.id = id;
this.val = val;
this.ruleType = ruleT
浏览 3提问于2022-05-06得票数 0
1回答
我正在使用flink siddhi并在处理大型对象时走出内存错误。在siddhi生成的输出流中，我有超过200个字段的对象，之后我有一些操作符来处理这个对象。flink版本1.7.2
java.lang.OutOfMemoryError: Java heap space
at com.esotericsoftware.kryo.io.Input.readBytes(Input.java:307)
at com.esotericsoftware.kryo.serializers.DefaultArraySerializers$ByteArraySerializer.read(De
浏览 12提问于2021-12-09得票数 0
3回答
我使用flink从kafka读取数据并将其转换为protobuf。我面临的问题是，当我运行java应用程序时，会出现以下错误。如果我将unknownFields变量名修改为其他内容，它就能工作，但是很难对所有protobuf类进行这种更改。
我还试图在阅读kafka时直接反序列化，但我不确定getProducedType()方法应该返回什么。
public static class ProtoDeserializer implements DeserializationSchema{
@Override
public TypeInformation getProduc
浏览 2提问于2016-07-09得票数 2
我在我的类上收到了一个CSV，我需要得到值来缓存一个POJO。我不需要将"file.csv“打开到目录中，逗号分隔的元素通过Flink传递给EventDeserializationSchema，这个元素用于"Event”来处理每个事件。
这里有一个例子：
在：“亚当”，“史密斯”，66岁，……‘12:01:00.000’“-Smith=‘Smith 3’>出:波乔
要做到这一点，我使用：
这是我的活动课，它应该能做到这一点，实际上目前还没有做任何事情。
import java.io.Serializable;
import org.apache.flink.shaded.
浏览 0提问于2020-10-29得票数 0
回答已采纳
这与Flink 1.13.2一起运行在Amazon的环境中。
此应用程序运行在Kafka主题上。当主题具有较小的流量时，这个应用程序运行良好，如果有更大的卷，我就会得到这个错误。如何进行故障排除、调优和修复？
我看到类似的问题，但这显然是Flink：的旧版本中的一个单独的问题。
例外跟踪是：
2021-12-30 18:16:45
java.util.concurrent.CompletionException: org.apache.flink.runtime.jobmanager.scheduler.NoResourceAvailableException: Could not acqui
浏览 46提问于2021-12-31得票数 6
回答已采纳
我正在使用Flink 1.13.1，并试图用10k记录的RPS(每秒速率)将数据写入kafka。我有一个由30个经纪人组成的kafka集群，我的flink工作做了一个过滤器操作员，只将数据汇到kafka。下面是我的生产者设置，最初我有5个接收器主题分区，但现在10个，我仍然得到同样的问题。另外，我尝试将request.timeout.ms设置为1分钟，而不是卡夫卡默认的30秒，但仍然得到了120001 ms has passed since batch creation。由于此错误，检查点失败。检查点大小仅为107 is，因为flink只向kafka提交偏移。作业并行性为36。这是纳卡的属性和
浏览 23提问于2022-04-11得票数 1
1回答
我在flink上有个问题。我实时计算引擎使用groovy脚本扩展计算类型(如:sum、average、、count等)。我们定义了一个标准的计算接口(AbstractCompute)，如果我想在这个框架中扩展一个计算类型，我只需要执行AbstractCompute.and，然后将groovy脚本存储在数据库中。然后，应用程序可以逐个任务地读取脚本，并通过GroovyClassLoader加载到虚拟机中。这个程序不用再用Flink了，看工作做的很好。原因是Flink使用另一个ClassLoader (FlinkUserCodeClassLoaders$ChildFirstClassLoader)
浏览 2提问于2019-01-14得票数 2
我正在使用Flink，我正在使用Kafka连接器。我从flink收到的消息是一个逗号分隔的项目列表。"'a'，'b'，‘c’，1，0.1....‘12:01:00.000’“其中一个包含事件时间，我想将此事件时间用于每个分区的水印(在kafka源代码中)，然后使用此事件时间进行会话窗口。我的情况与往常有点不同，因为据我所知，人们通常使用"kafka时间戳“和SimpleStringSchema()。在我的例子中，我必须编写自己的反序列化程序来实现DeserializationSchema并返回一个元组或Pojo。所以基本上用我自己的函数替换Si
浏览 69提问于2020-10-24得票数 1
回答已采纳
基本上，我希望将Flink自定义JAR文件部署到新的AWS EMR集群中。这是我所做的事情的总结。我创建了一个新的AWS EMR集群。
步骤1:软件和步骤更改-
- Created a AWS EMR cluster with flink as the service. (EMR release version - 5.17.0) and clicked Flink 1.5.2 as the software configuration.
- Entered the Configuration JSON:- {“分类”："flink-conf“、”属性“：{ "jo
浏览 0提问于2018-10-09得票数 0
回答已采纳
1回答
我有一个在生产环境中运行的Flink流作业，我需要对主转换代码进行更改。
生产中的代码实际上如下所示：
stream
.filter(inboundData -> inboundData.hasToBeFiltered())
.uid("filtered-data")
.keyBy(data -> data.getMyStringKey())
.process(doSomething())
.uid("processed-inbound-data-id");
我需要改变keyBy()操作符使用inboundData PO
浏览 0提问于2019-08-08得票数 1
是否有人知道，在远程Flink集群上运行程序时，以下错误的根源是什么？
我该怎么解决呢？
org.apache.flink.client.program.ProgramInvocationException: The main method caused an error.
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:512)
at org.apache.flink.client.program.PackagedProgram.invokeInte
浏览 16提问于2015-12-10得票数 1
回答已采纳
我刚刚尝试使用Rosetta 2兼容层在苹果Mac上运行和新的M1处理器。
不幸的是，它在以下堆栈跟踪中失败：
flink-1.12.2 ./bin/flink run ./examples/streaming/WordCount.jar
Executing WordCount example with default input data set.
Use --input to specify file input.
Printing result to stdout. Use --output to specify output path.
WARNING: An illegal refl
浏览 2提问于2021-04-13得票数 3
2回答
我尝试在Flink 1.4上提交一个作业，并得到以下异常。
知道怎么解决这个问题吗？
Caused by: org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
at org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$6.apply$mcV$sp(JobManager.scala:897)
at org.apache.flink.runtime.jobmanag
浏览 4提问于2017-12-21得票数 6
回答已采纳
我在一个由三个节点组成的集群上执行Apache Flink程序。其中之一也可以作为jobmanager和taskmanager。另外两个只是taskmanager。
当我启动我的程序时(我在jobmanager上执行)，我得到了以下错误(在程序没有真正执行一分钟后)：
java.lang.Exception: TaskManager was lost/killed: c4211322e77548b791c70d466c138a49 @ giordano-2-2-100-1 (dataPort=37904)
at org.apache.flink.runtime.instance.SimpleS
浏览 0提问于2017-09-16得票数 1
我在Apache flink 1.3.2中尝试了几个Flink translate()函数:一个来自BatchTableEnvironment，另一个来自StreamTableEnvironment。程序给了我错误，我迷路了。下面的程序在运行时抛出异常(不知道它需要什么TypeInformation )。它只使用应与结果所需完全匹配的类型信息。但是它不工作，有什么帮助吗？我想我提供了所有的details...the
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.j
浏览 0提问于2017-10-29得票数 0
1回答
我尝试过Janino，通过将字符串传递给janino编译器，我能够执行一个动态类。
现在，我应该动态创建几个类，然后将它们导入到另一个动态类中。
举个例子：
从字符串编译A类
从字符串编译B类
创建类C，放在"import mypackage.A；import mypackage.B；等“中。
编译C类
我如何才能取得这样的结果？
我的第一节动态课程是：
package com.Flink.POJO;
public class RainPOJO {
private String Altitude;
private String City_
浏览 3提问于2015-09-30得票数 1
我是Flink流媒体的初学者。
使用RowCsvInputFormat读取文件时，Kryo序列化程序创建的代码不能正常工作。
代码在下面。
val readLocalCsvFile = new RowCsvInputFormat(
new Path("flink-test/000000_1"),
Array(Types.STRING, Types.STRING, Types.STRING),
"\n",
","
)
val read = env.readFile(
浏览 23提问于2022-01-06得票数 0
我们最近正在将flink集群升级到版本1.9.1。出现hadoop s3a相关错误。消息如下。 2020-01-16 08:39:49,283 ERROR org.apache.flink.runtime.blob.BlobServerConnection - PUT operation failed
org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.fs.UnsupportedFileSystemException: No FileSystem for scheme "file"
at o
浏览 44提问于2020-01-16得票数 0
Added Depedency Pom Details :
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_2.11</artifactId>
<version>1.7.1</version>
</dependency>
<depe
浏览 0提问于2019-04-04得票数 2
回答已采纳
1回答
我正在尝试将flink设置为从检查点恢复。在大多数情况下，这似乎是有效的，然而，在将其部署到我们的分段环境大约一周后，作业管理器已经开始崩溃循环，因为在尝试启动作业的“作业主机”时超时。 我使用在高可用性模式下部署的flink 1.7.2和zookeeper 3.4.9-1757313，只是为了方便检查点恢复。我在kubernetes上只部署了一个作业管理器作为有状态集。一定是什么原因导致服务器崩溃，并且在重新启动时，启动(可能)恢复的作业的作业主机的代码似乎失败。 我以前见过一次，清除所有的flink zookeeper条目( zk中的rmr /flink)，然后重新启动flink集群“修
浏览 57提问于2019-06-19得票数 1
我正在尝试将我们的工作从flink 1.4.2升级到1.7.1，但是在提交作业后，我一直遇到超时。flink作业在我们的hadoop集群上运行(版本2.7)和Yarn。
我看到了以下行为：
使用与我们在1.4.2: 1.5.6 / 1.6.3 / 1.7.1中使用的相同的flink- works. same，当1.4.2起作用时，所有版本都超时。
使用1.5.6与“模式:遗留”(关闭翻转-6)工作
使用1.7.1与"mode: legacy“一起使用会导致超时(我假设此选项已被删除，但文档已过时？)
当超时发生时，我得到以下堆栈跟踪：
INFO class j
浏览 1提问于2019-02-19得票数 1
回答已采纳
我从官方的flink存储库使用以下docker-compose.yml启动了flink。我只添加了到外部hadoop网络的连接。 version: "2.1"
networks:
hadoop:
external:
name: flink_hadoop
services:
jobmanager:
image: flink:1.7.1-hadoop27-scala_2.11
container_name: flink-jobmanager
domainname: hadoop
networks:
- h
浏览 1293提问于2019-01-04得票数 2
回答已采纳
我正在尝试修改一个case类，它有大约240个变量，其中一些是我定义的其他pojos : Signal (case类)有变量PowerPojoFeature，UserFeature。当我修改任何pojo添加新变量时，它会抛出exception rg.apache.flink.util.StateMigrationException下面:操作符typeSerializer的新状态state不能是不兼容的。在org.apache.flink.runtime.state.DefaultOperatorStateBackend.getListState(DefaultOperatorStateBac
浏览 5提问于2019-10-18得票数 1
2回答
只是尝试从flink 1.3迁移到1.4，并在linux机器上遇到此异常：(不能在windows上重现)。
我还导入了这个包：
// https://mvnrepository.com/artifact/org.apache.flink/flink-shaded-hadoop2
compile group: 'org.apache.flink', name: 'flink-shaded-hadoop2', version: '1.4.0'
有什么帮助吗？
在flink控制台：
TriggerWindow(TumblingProcessingTim
浏览 22提问于2018-02-13得票数 1
我有一个带有lombok注释的POJO。我的项目创建了一个胖jar，以部署在由apache气流组织的工作流中，并作为flink作业执行。
部署工作正常，在运行时，我在lombok构建器调用上得到一个noclassdeffounderror。
java.lang.NoClassDefFoundError: com.Pojo$PojoBuilder
at com.Pojo.builder(Pojo.java:10)
我检查了机器上解压缩的胖罐子，pojo类有构建器。你知道是什么导致了这个问题吗？
由于公司的规定，可能无法共享准确的代码。
浏览 1提问于2022-05-30得票数 0
我希望使用架构信息将DataStream转换为DataStream
输入
参数DataStream
{"fields":["China","Beijing"]}
args1模式
message spark_schema {
optional binary country (UTF8);
optional binary city (UTF8);
}
期望输出
{"country":"china", "city":"beijing"}
我的代码如下所示
public DataSt
浏览 19提问于2020-03-07得票数 0
更新
为了提供一些上下文，使用LZ4压缩启用了Kafka主题。
当识别压缩类型时，尝试将LZ4与运行时由于某种原因可以找到的KafkaLZ4BlockInputStream相关联。
@Override
public InputStream wrapForInput(ByteBuffer inputBuffer, byte messageVersion, BufferSupplier decompressionBufferSupplier) {
try {
return new KafkaLZ4BlockInputStream(inputBuffer, decompres
浏览 18提问于2021-12-09得票数 1
我们有一个用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
浏览 16提问于2019-03-21得票数 1
2回答
最近，apache的稳定版本(1.6.2)发布。我读过。但是，当我运行以下命令时：
./bin/flink run examples/streaming/SocketWindowWordCount.jar --port 9000
我得到以下错误：
The program finished with the following exception:
org.apache.flink.client.program.ProgramInvocationException: Job failed. (JobID: 264564a337d4c6705bde681b34010d28)
at or
浏览 0提问于2018-11-17得票数 0
回答已采纳
我尝试运行wordcount示例：../bin/flink run WordCount.jar
并在执行几分钟后给我一个“与JobManager通信失败的错误”。
来源：
异常情况如下：
Executing WordCount example with built-in default data.
Provide parameters to read input data from a file.
Usage: WordCount <text path> <result path>
org.apache.flink.client.program.ProgramI
浏览 1提问于2015-12-17得票数 3
我正在尝试创建到oracle数据库的Flink JBDC接收器。当在本地运行时(从junit测试和minicluster)，它可以工作，但是当部署在k8s中时，它会抛出一个异常，说它找不到合适的驱动程序。Classpath是：
Classpath: /flink/lib/flink-cep-scala_2.12-1.13.5-stream1.jar:/flink/lib/flink-connector-jdbc_2.12-1.13.5.jar:/flink/lib/flink-csv-1.13.5-stream1.jar:/flink/lib/flink-json-1.13.5-stream1
浏览 16提问于2022-05-24得票数 1
回答已采纳
在Flink中，使用readCsvFile解析CSV文件会在包含引号的字段中引发异常，如"Fazenda São José ""OB"" Airport"：
org.apache.flink.api.common.io.ParseException: Line could not be parsed: '191,"SDOB","small_airport","Fazenda São José ""OB"" Airport",-21.42519950866699
浏览 4提问于2016-07-11得票数 4
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
云直播
对象存储活动推荐
腾讯云开发者
