我正在k8s集群上运行Flink 1.11,在尝试更新log4j-console.properties文件时遇到以下错误: Starting Task Manager Enabling required built-in plugins Linking flink-s3-fs-hadoop-...
我是一名新的flink用户,遇到了以下问题。我在YARN集群上使用flink将从RDBMS中提取的相关数据传输到HBase。我用Java编写了flink批处理应用程序,并使用多个ExecutionEnvironments(每个环境对应一个RDB表格),以便按顺序逐个传输表格(因为调用env.e...
首先,这与当我重新运行Flink消费者时,Kafka再次消耗最新消息非常相似,但并不完全相同。那个问题的答案似乎不能解决我的问题。如果我在那个答案中错过了什么,请重新表述答案,因为我显然错过了某些内容。 虽然问题是完全相同的-- Flink(kafka连接器)重新运行它关闭之前看到的最后3-...
我正在阅读有关kafka stream中连接限制的内容,例如KTable KTable非关键字联接或KTable GlobalKTable等。 我发现Flink似乎支持所有这些功能。从我所读的内容来看,动态表格听起来像是一个KTable。 我想知道首先它们是否是相同的概念,然后Flink如...
我正在尝试使用Apache Flink 1.11创建一个源表,以便可以访问JSON消息中的嵌套属性。我可以获取根属性的值,但不确定如何访问嵌套对象。 文档建议应该使用MAP类型,但是当我设置它时,出现以下错误: : java.lang.UnsupportedOperationExcepti...
我刚刚拿到了下面这个例子用于并行处理,并有一些相关的问题: setParallelism(5) 只为 sum 设置 Parallelism 5,还是对 flatMap 和 sum 都设置了 Parallelism 5? 我们是否可以将不同的并行度分别设置给不同的操作符,例如分别设置 sum...
我希望得到一些简单的规则或指导方针,以确定operator或job并行度应设置为什么值。对我来说,它似乎应该是一个小于等于可用任务槽数量的数字? 例如,假设我有两台task manager机器,每台机器上有4个任务槽。假设在集群上没有其他作业正在运行,在像filter和map这样的操作中,我...
我正在尝试将Apache Flink与AWS Kinesis一起使用。 文档说我必须自己构建连接器。 因此,我构建了连接器,并为我的项目添加了jar文件,同时在pom.xml文件中添加了依赖项。<dependency> <groupId>org.apache....
我有一个使用bzip2压缩的维基百科转储文件(从http://dumps.wikimedia.org/enwiki/下载),但我不想解压它:我想在解压时进行处理。 我知道可以在普通Java中完成此操作(例如参见Java - Read BZ2 file and uncompress/parse ...
在Apache Flink中,“JoinFunction”和“CoGroupFunction”有什么区别?它们的语义和执行方式有何不同?