我已经使用flink consumer和producer概念创建了一个Flink应用程序/作业。 现在我还有一个需求，比如我需要创建一个Producer应用程序，该应用程序将数据发布到在Flink应用程序之上使用的topic。我使用KafkaProducer创建了这个Producer应用程序，它可以正常工作。但我还需要将此Producer应用程序作为Flink作业运行。我做了谷歌搜索，我无法找到细节，以创建一个Flink应用程序只与生产者。
浏览 5提问于2020-07-12得票数 1
我想在Apache Flink中做流式作业，在Apache Flink(Scala)中做Kafka -> Flink -> HIVE。谁能给代码样本，因为他们的官方文件不是很清楚。 这应该是流过程。
浏览 11提问于2020-07-09得票数 1
回答已采纳
我已经写了一个非常简单的Flink流作业，从卡夫卡使用FlinkKafkaConsumer082的数据。
protected DataStream<String> getKafkaStream(StreamExecutionEnvironment env, String topic) {
Properties result = new Properties();
result.put("bootstrap.servers", getBrokerUrl());
result.put("zookeeper.connect", ge
浏览 0提问于2015-11-03得票数 2
背景
我们有一个Kafka主题，有稳定的数据流。为了处理它，我们有一个无状态的Flink管道，它使用该主题并写入另一个主题。
时不时地，我们的Flink不能处理突发信息。我们不想配置我们的Flink流水线和集群来始终支持我们可以拥有的最大负载，我们希望根据负载动态扩展。(预算原因$$$)
我们想到的解决方案
为此，一种方法是向Flink集群添加/删除节点，并更改Flink管道操作符的并行度。这将需要使用快照停止Flink作业，重新配置并行度，然后使用新的并行度重新启动。
这将是伟大的，但我们不能允许自己的停机时间，它产生。我们必须在不停机的情况下扩大/缩小规模。
如果我们使用常规的Kafka用
浏览 27提问于2019-11-14得票数 1
我正在尝试向Kubernates中部署的ververica平台提交作业，但我收到了下面的消息，我向Flink standalone提交了相同的代码，它的工作正常！！我使用的是Flink 1.10.1，代码是Scala 2.12。 Caused by: org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Could not find a suitable table factory for 'org.apache.flink.table.factorie
浏览 74提问于2020-07-12得票数 1
回答已采纳
我们有一些在纱线上运行的flink作业。我们希望上传flink作业日志到ELK，以简化调试/分析。目前，flink任务管理器将日志写入/mnt/flinklogs/$application_id/$container_id。我们想让它写入一个没有$applicatoin_id/$container_id嵌套结构的目录。 我和env.log.dir: /mnt/flink试过了。使用此设置时，配置不会正确传递。 -Dlog.file=/mnt/flinklogs/application_1560449379756_1312/\
container_e02_1560449379756_
浏览 72提问于2019-08-27得票数 0
我计划从Flink 1.5.2升级到1.6.0，然后进行作业迁移。为了最大限度地减少作业的暂停时间，我计划同时运行两个Flink集群，在成功迁移作业后，我会停止旧的。但是，当我尝试通过运行目录Flink1.5.2/bin中的stop-cluster.sh来停止Flink集群时，我发现停止的集群是Flink 1.6.0，而不是预期的Flink 1.5.2。
我做了一些测试，发现stop-cluster.sh只是停止最近启动的Flink集群，也就是说，如果你先启动集群1.6.0，然后启动Flink 1.5.2，之后当你运行stop-cluster.sh时，即使你在集群1.6.0目录Flink1.6
浏览 23提问于2018-08-30得票数 1
回答已采纳
我有多个Kafka主题(多租户)，并且我根据主题的数量多次运行相同的作业，每个作业消耗来自一个主题的消息。我已经将文件系统配置为状态后端。 假设有3个作业在运行。检查点在这里是如何工作的？这3个作业是否都将检查点信息存储在同一路径中？如果任何作业失败，该作业如何知道从何处恢复检查点信息？我们过去常常在向flink群集提交作业时提供作业名称。这跟它有什么关系吗？一般而言，Flink如何区分作业及其检查点信息，以便在作业出现故障或手动重新启动作业时进行恢复(无论作业相同还是不同)？ Case1:如果作业失败，会发生什么情况？ Case2:如果我们手动重新启动作业会发生什么？ 谢谢
浏览 33提问于2020-07-16得票数 0
回答已采纳
我有一个在flink流中序列化的参数params， class P extend Serializable {...}
val params = new P(...)
val env = StreamExecutionEnvironment.getExecutionEnvironment
env.addSource(new MySource(params))
.map(new MyMap(params))
.addSink(new MySink(params))
env.setParallelism(1)
env.execute("My Job") 但是驱动节点
浏览 24提问于2020-06-30得票数 0
回答已采纳
假设我想用Flink实现一个ETL作业，源和接收器都是Kafka主题，只有一个分区。
源和接收器中记录的顺序关系到下游(我的ETL中有更多的作业消耗接收器，作业由其他团队维护)。
有没有办法确保记录在接收器中的顺序与源相同，并使并行性超过1？
浏览 2提问于2021-10-28得票数 0
回答已采纳
1回答
我在一台机器上启动flink (bin/ start -cluster.sh)，并通过flink web UI提交作业。如果作业有问题，比如接收器mysql表不存在或keyby字段错误，不仅这个作业失败，我不得不取消失败的任务，但在取消后，任务管理器似乎被“杀死”，它在flink web ui中消失。
有没有容错的解决方案(taskmanager会被失败的作业杀死)？唯一的方法是在纱线上运行flink？
浏览 19提问于2018-09-04得票数 0
我们使用Flink Table API在Flink应用程序中使用Kafka主题。
当我们第一次提交应用程序时，我们首先从自定义注册表中读取最新的模式。然后使用Avro模式创建Kafka数据流和表。我的数据序列化程序的实现的工作方式类似于Confluent模式注册表，它先检查模式ID，然后使用注册表。因此，我们可以在运行时应用正确的模式。
但是，我不知道如何在不重新部署作业的情况下更新表模式和重新执行SQL。有没有办法让后台线程检查模式更改，如果有更改，则暂停当前执行，更新表模式并执行SQL。
这对于连续交付对应用程序的模式更改特别有用。我们已经进行了兼容性检查。
浏览 2提问于2020-11-14得票数 0
1回答
我需要为我的项目设置一个流环境。下一个是dataset：
我计划使用PostgreSQL和Apache作为源代码，然后用Apache或Flink连接到Kafka，但是问题是数据集大约是10000个文本文件。
因此，问题是将~10000文件(每500行左右)导入到一个表中的最有效/最优雅的方法是什么。
目前，我已经尝试过使用psycopg2和executemany方法(超级慢)和Apache连接器(速度很好，但不知道如何在一个作业中迭代10000多个文件)的python脚本。
浏览 1提问于2016-06-03得票数 1
回答已采纳
我正试图理解和管理Flink对本地存储的使用，就像在我的用例中一样，我需要确保没有加密就不会存储任何数据。
阅读这些文档我可以看到
如果状态变得太大，无法保存在内存中，RocksDb可以使用本地存储。
Flink使用本地fs，以防存储在内存中的数据变得太大(e在进行大型聚合活动的批处理作业中)
在使用本地存储时，是否还有其他需要考虑的情况？以上两种情况下的文件夹似乎都是由属性taskmanager.tmp.dirs驱动的。
如果我想为包括加密在内的上述情况编写特定的数据序列化程序/写入程序，是否有人可以指出我应该查看哪个类？
浏览 2提问于2018-02-12得票数 1
回答已采纳
我使用Kubernetes (Openshift)部署许多微服务。我希望使用同样的方法来部署我的一些Flink作业。Flink作业是关键的-有些作业是无状态的，它们处理每一个数据(准确地说是一次)，有些作业是有状态的，可以在流中查找模式或对时间作出反应。没有作业能够容忍长时间的停机或频繁的关闭(因为编程错误，Flink退出的方式)。
我发现文档大多倾向于将k8s中的Flink作业部署为Job Cluster。但是，我们应该如何采取一种实际的做法来做到这一点呢？
虽然k8s可以重新启动失败的pod，但是Flink如何恢复其状态以恢复？
Flink pod可以复制多个吗？当存在两个或多个豆
浏览 0提问于2020-02-13得票数 3
1回答
我使用Apache Flink将来自给定的Kafka主题集的更新传播到Elasticsearch集群中。
我面临的问题是，有时Elasticsearch集群会进化，而我必须通过(1)修改映射，当我不得不将Flink作业指向新的别名/索引时，(2)复制到data...and时，就会有大量的更新到旧的索引。
所以我想知道什么是最好的方法来解决这个问题。我可以休息，但如果可能的话，我想避免这样做。我试图使Flink作业减速或暂停(Kafka)输入源，直到迁移完成，但我没有找到任何端点。
Flink作业在应用程序模式下运行。
如果有人能够解释一下如何做到这一点:通过API或其他类似工具暂停/恢复作
浏览 5提问于2022-06-21得票数 0
我正在开发一个项目，该项目使用flink (Version - 1.4.2)将大量数据摄取到my图形数据库 (Janusgraph)。数据摄取有两个阶段，一个是顶点数据的摄取，另一个阶段是对图db的边缘数据的摄取。顶点数据摄取运行时没有任何问题，但是在边缘摄入期间，我遇到一个错误，即与任务管理器taskmanagerName失去了连接。来自flink-taskmanager-b6f46f6c8-fgtlw的详细错误跟踪附在下面：
2019-08-01 18:13:26,025 ERROR org.apache.flink.runtime.operators.BatchTask
- Err
浏览 1提问于2019-08-04得票数 0
我有兴趣了解Flink如何更好地执行检查点(或保存点)。以下是我的好奇之处：
Flink作业管理器是否从任务管理器收集所有状态，然后将这些数据推送到深度存储服务(如GCP或S3)？或者，每个任务经理和职务经理是否相互独立地推动状态。
浏览 4提问于2022-04-30得票数 0
回答已采纳
在KDA (kakfa --> flink(beam) --> ElasticSearch)上运行一个闪烁光束作业这个简单的作业不会在KDA上启动并进入无限循环。AWS KDA支持回复说，作业在启动期间读取记录，这是失败的原因。 该应用程序的停靠版本在kubernetes中的3个任务管理器上运行得很流畅，但在KDA上则不是。因为KDA有2分钟的超时时间来启动作业。 根据我的理解，一旦工作开始，Flink就开始读取记录，我如何将启动时间减少到2分钟以内，因为工作是非常基本的读取kafka和存储到ES的记录。
浏览 25提问于2021-02-09得票数 0
回答已采纳
我正在使用flink从Azure数据湖中读取数据。但flink无法找到Azure数据湖文件系统。我已经实现了spark来读取Azure Data Lake文件系统文件。
因此，我尝试从flink中触发火花，从Azure数据湖中读取数据，并执行一些逻辑操作，然后将列表返回给flink。所以flink可以使用这些数据？
是否可以从flink触发spark作业？或者如何配置flink来理解Azure数据湖文件系统。有人能在这方面给我指点一下吗？
浏览 2提问于2017-07-11得票数 0
使用来自Kafka主题的数据，Flink和SparkStreaming都提供了检查点机制，前提是auto.commit.enabled设置为false。火花博士说：
火花输出操作至少是一次。因此，如果您想要完全等同于一次语义，则必须在幂等输出之后存储偏移量，或者在原子事务中存储偏移量与输出一起。
但是Flink文档说：
在启用了Flink的检查点之后，Flink Kafka消费者将使用来自某个主题的记录，并定期检查其所有Kafka偏移量，以及其他操作的状态，以一致的方式。如果作业失败，Flink将恢复流程序到最新检查点的状态，并重新使用Kafka的记录，从存储在检查点中的偏移量开始。
浏览 6提问于2017-12-12得票数 1
2回答
我们正在部署一个新的Flink流处理作业，它的状态(存储)需要使用历史数据进行初始化，并且在开始处理任何新的应用程序事件之前，这些数据应该在状态存储中可用。我们不希望大幅修改Flink作业来加载历史数据。我们考虑编写另一个单独的Flink作业来处理历史数据，更新它的状态存储并创建一个保存点，并使用这个保存点来初始化主Flink作业中的状态。看起来状态处理器应用程序接口只能与DataSet应用程序接口一起使用，并且想知道是否有其他解决方案。谢谢。
浏览 47提问于2021-04-23得票数 0
我有一个Flink流程序，从卡夫卡的主题读取数据。在程序中，auto.offset.reset被设置为“最小”。当在IDE/Intellij-IDEA中测试时，程序总是可以从主题的开头读取数据。然后我建立了一个flink/kafka集群，并将一些数据生成到kafka主题中。当我第一次运行流式作业时，它可以从主题的开头读取数据。但在此之后，我停止了流作业，并再次运行它，它将不会从主题的开头读取数据。如何才能使程序始终从主题的开头读取数据？
Properties properties = new Properties();
properties.put("bootstrap
浏览 5提问于2015-12-15得票数 0
2回答
如何在Flink转换中更新配置？
、、、、
给定一个Flink流作业，该作业将map()操作应用于流。
此map()操作从某些属性读取其配置，并相应地映射数据。例如，配置指定读取属性"input"，并使用不同的属性名"output“将其写入流。这个已经很好了。
现在，配置发生了变化，例如，转换将为输出使用不同的属性名。
因此，我正在寻找一种方式，让所有的Flink任务重新阅读一个新的配置在运行时。
有没有可能
将KafkaSourcewait挂起，直到管道耗尽(flush)trigger集群中的所有任务，以重新读取配置文件(coordinated)resume (coordinated)resume KafkaS
浏览 13提问于2022-10-11得票数 0
1回答
Apache通过从检查点恢复作业来保证在故障和恢复时处理一次，检查点是分布式数据流和操作符状态的一致快照(Chandy-Lamport算法用于分布式快照)。此保证在故障转移时恰好是一次。
在正常的集群操作中，Flink如何保证只处理一次，例如给定从外部源读取的Flink源(例如Kafka)，Flink如何保证事件从源读取一次？事件源和Flink源之间是否存在某种类型的应用程序级别？另外，Flink如何保证事件准确地从上游运营商到下游运营商之间只发生一次？对于接收到的事件，这也需要任何形式的附加条件吗？
浏览 3提问于2020-09-30得票数 3
回答已采纳
使用flink版本1.13.1
我写了一个自定义的度量报告，但在我的flink中似乎不起作用。启动flink时，JobManager显示警告日志，如下所示：
2021-08-25 14:54:06,243 WARN org.apache.flink.runtime.metrics.ReporterSetup [] - The reporter factory (org.apache.flink.metrics.kafka.KafkaReporterFactory) could not be found for reporter kafka. Available f
浏览 10提问于2021-08-25得票数 2
回答已采纳
1回答
我的flink作业的一个参数是动态的，我有一个api来获取动态值。是否可以每次都在源端调用接口，根据参数拉取数据？这是正确的方式吗？这会在flink作业中造成任何麻烦吗？
浏览 0提问于2021-05-27得票数 0
我需要我的Flink批处理作业从我的Spring Boot项目的postgres数据库中读取数据。我在Flink中看到的选项仅用于读取csv文件。在这种情况下，我需要从我的表创建一个csv文件。有什么解决方案吗？
浏览 36提问于2020-07-23得票数 2
回答已采纳
1回答
了解flink保存点和检查点
、、、、
考虑具有如下管道的Apache流应用程序：
Kafka-Source -> flatMap 1 -> flatMap 2 -> flatMap 3 -> Kafka-Sink
其中，每个flatMap函数都是一个无状态运算符(例如，Datastream的正常.flatMap函数)。
当传入消息在flatMap 3上挂起时，检查点/保存点是如何工作的？消息是从flatMap 1开始重新启动后重新处理，还是跳到flatMap 3
我有点困惑，因为似乎将应用程序状态称为我可以在有状态运算符中使用的东西，但我的应用程序中没有有状态运算符。是否保存和恢复了“处理过程”，还是在失败
浏览 2提问于2019-03-04得票数 0
1回答
当我的制作人把所有的消息都流到Kafka之后，在Flink完成处理之后，我希望能够终止Flink作业，这样它就不会继续运行了，这样我就可以知道Flink什么时候完成了所有的数据处理。我也不能使用批处理，因为我需要Flink运行平行于我的Kafka流。
通常，Flink在一个isEndOfStream类中使用DeserializationSchema方法来查看它是否应该提前结束(返回方法中的true将自动结束作业)。但是，当使用Flink作为来源时，新的KafkaSource类不赞成在反序列化器中使用isEndOfStream方法，不再检查它是否应该结束流。还有其他方法可以提前终止Flink工作
浏览 13提问于2022-10-03得票数 0
我正在为一个大容量流用例(每秒数以千万计的事件)开发一个DataStream-based Flink应用程序。这些数据是从卡夫卡主题中消耗的，并且已经根据特定的键进行了切分。我的意图是在Flink端创建特定于键的状态，以运行自定义分析。我无法解决的主要问题是，如何创建键控状态，而不对keyBy()强加的传入数据进行重新配置。
我可以保证Flink作业的最大并行性将小于或等于源Kafka主题中的分区数，因此逻辑上没有必要进行洗牌。 StackOverflow问题的答案表明，可能有可能以一种符合Flink预期的方式将数据写入Kafka，然后使用reinterpretAsKeyedStream()。
浏览 4提问于2022-08-08得票数 2
回答已采纳
当我们在作为SQL Server数据库表的源的Flink SQL CLI中选择任何表时，为什么Flink作业在从数据库表中获取所有记录后“完成”？有没有可能让它“运行”，这样它就可以从数据库表中获取最新的数据？
浏览 4提问于2020-12-25得票数 1
我正在测试Flink 1.3.0中的弹性特性。我有一个具有检查点启用和固定延迟重启策略的作业.当我杀死一个TaskManager JVM时，一段时间后，作业将在剩余的节点上正确地重新启动。但是，当我添加一个新节点时，作业不会自动重新启动以利用它。
我试着使用bin/flink stop <jobId>，但它总是给我java.lang.IllegalStateException: Job with ID <jobId> is not stoppable.
如何重新启动作业以利用附加节点？
浏览 2提问于2017-06-19得票数 1
回答已采纳
1回答
我从我的.jar文件运行多个作业。我想在我的工作之间共享状态。但是所有输入在每个作业中都会消耗(来自kafka)，并生成重复的输出。我看到了flink面板。所有的工作‘记录发送’是3。我想必须拆分数字到我的工作。 我使用此命令创建作业 bin/flink run app.jar 我怎么才能修复它？
浏览 46提问于2019-09-29得票数 0
回答已采纳
1回答
为了学习目的，我正在构建一个简单的数据管道。我有来自Kafka的实时数据，我想使用Flink进行一些转换。
不幸的是，我不确定我是否正确地理解了部署选项。在Flink文档中，我找到了关于Docker组成和应用模式的部分。它说，我只能将一个作业部署到Flink：
Flink应用程序群集是运行单个作业的专用群集。在这种情况下，您可以将作业部署到集群中，这样就不需要额外的作业提交了。
作业工件包含在容器内Flink的JVM进程的类路径中，包括：
您的作业jar，您通常会将其提交到会话集群和
所有其他必要的依赖关系或资源，不包括在Flink中。
要使用Docker为单个作业部
浏览 5提问于2022-01-08得票数 0
回答已采纳
我正在编写一个测试flink两步提交的案例，下面是概述。
sink kafka正是卡夫卡的制作人。sink step是mysql接收器扩展two step commit。sink compare是mysql接收器扩展two step commit，这个接收器偶尔会抛出一个exeption来模拟检查点失败。
当检查点失败并恢复时，我发现mysql两步提交将运行良好，但是kafka消费者将读取与上一次成功的抵消，甚至在这个检查点失败的之前他也会生成消息。
在这种情况下，如何避免重复消息？
谢谢你帮忙。
env
flink 1.9.1
java 1.8
kafka 2.11
浏览 6提问于2019-12-05得票数 1
回答已采纳
我正在构建一个要求低于要求的应用程序，我刚刚开始使用flink。
摄入卡夫卡的数据，比如50个分区(输入速率- 100,000毫希/秒)
从Kafka读取数据并对每一数据进行实时处理(进行一些计算、与旧数据的比较等)
将输出存储在Cassandra上
我正在寻找一个实时流媒体平台，并发现Flink是一个非常适合实时和批处理。
您认为flink最适合我的用例吗?还是应该使用Storm、Spark流或任何其他流平台？
我是否需要在google数据流中编写数据管道来执行flink上的步骤序列，还是有其他方法来执行实时流的步骤序列？
假设我的每一次计算花费大约20毫秒，
浏览 2提问于2016-06-10得票数 3
我有多个flink作业，它有相同的输入源，kafka主题和输出格式也是一样的。
源-> flink作业1 ->输出
源-> flink作业2 ->输出
源-> flink作业3 ->输出
源-> flink作业4 ->输出
。
。
。
源-> flink作业n ->输出
输出格式类似于Object(pk: String, variable1: String, variable2: Boolean)
我想要使用所有的输出，并进行组合输出，比如输出数组的json
最终所需输出(pk: String, variable1: List[Str
浏览 4提问于2022-09-20得票数 0
回答已采纳
我正在运行带有3个m5.x大节点(1个主节点，2个核心)和Flink 1.8的EMR集群(emr-5.24.1)。
在主节点上，我使用以下命令在纱线集群中启动Flink会话：
flink-yarn-session -s 4 -jm 12288m -tm 12288m
这是每个TaskManager的最大内存和插槽，让我根据所选的实例类型设置它。
在启动期间，有一个日志：
org.apache.flink.yarn.AbstractYarnClusterDescriptor - Cluster specification: ClusterSpecification{maste
浏览 0提问于2019-07-08得票数 0
回答已采纳
我有一个在AWS EMR中运行的高度并行(400)的Flink应用程序。它使用BucketingSink源码Kafka并汇入S3 (使用RocksDb后端设置检查点)。目的地使用"s3a://“前缀定义。Flink作业是一个持续运行的流媒体应用程序。在任何给定的时间，所有工作进程加在一起都有可能生成/写入400个文件(由于400个并行度)。几天后，其中一个工人将失败，并出现以下异常：
org.apache.hadoop.fs.s3a.AWSS3IOException: copyFile(bucket/2018-09-01/05/_file-10-1.gz.in-progress, bu
浏览 4提问于2018-12-05得票数 0
我正在探索使用Flink rocksDb状态后端，文档似乎意味着我可以使用常规的文件系统，比如：file:///data/flink/checkpoints，但是代码javadoc在这里只提到了hdfs或s3选项。
我想知道是否可以使用本地文件系统与flink rocksdb后端，谢谢！
Flink文档：
Flink代码：
浏览 3提问于2019-10-29得票数 3
回答已采纳
我一直试图在GCP ()上使用Flink，但谷歌云存储访问存在一个问题。
所以，我遵循了这里解释的步骤()
所以，我创造了一个码头形象，比如；
ARG GCS_CONNECTOR_VERSION=latest-hadoop2
ARG FLINK_HADOOP_VERSION=2.8.3-10.0
ARG GCS_CONNECTOR_NAME=gcs-connector-${GCS_CONNECTOR_VERSION}.jar
ARG GCS_CONNECTOR_URI=https://storage.googleapis.com/hadoop-lib/gcs/${GCS_CONNECTOR_
浏览 12提问于2022-08-03得票数 1
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
1回答
我试图通过提交给Flink中的作业管理器来运行java中的一个基本程序。我有一个来自open CV的本地库。当我尝试提交作业时，我得到"java.lang.UnsatisfiedLinkError: no opencv_java310 in java.library.path"，但是当我通过设置flink执行环境在eclipse上运行它时，我得到了正确的结果。我遵循了apache flink支持网站上的一些解决方案：，并相应地修改了我的conf.yaml文件(通过指向env.java.opts：-Djava.library.path="/path of Open CV
浏览 0提问于2016-05-23得票数 1
3回答
我正在flink群集内运行flink流作业。我需要模拟一个作业失败的场景。我在我的源代码中引入了一个损坏的事件。当该事件发生时，我看到抛出异常的作业和任务已经从检查点重新启动。我已经设置了以下配置： restart-strategy.fixed-delay.attempts: 1
restart-strategy.fixed-delay.delay: 5 s 但这些配置并没有得到尊重。即使在第一次尝试后，任务仍在恢复。根据文档，作业应在第一次尝试后失败。我需要模拟作业失败的场景。该怎么做呢？
浏览 38提问于2020-10-13得票数 0
回答已采纳
当打包使用flink连接器和第三方库(用于处理)的flink作业时，哪些依赖项应该在jar中结束，这样就可以使用("flink run jarfile")在flink集群中启动它？
做一个胖罐子是你想要的方法吗？
如果用scala编写作业，是否在jar中包含scala默认库？
我没有找到任何有关如何打包flink的作业的文档。
浏览 1提问于2016-06-17得票数 0
回答已采纳
2回答
使用flink 0.10.1在本地，由于以下错误，我无法与作业管理器连接：
Association with remote system [akka.tcp://flink@127.0.0.1:49789] has failed, address is now gated for [5000] ms. Reason is: [scala.Option; local class incompatible: stream classdesc serialVersionUID = -2062608324514658839, local class serialVersionUID = -114
浏览 3提问于2015-12-30得票数 0
回答已采纳
1回答
让我把它说出来，我是一个非常初学者的Flink，并试图抓住尽可能多的概念。
比方说，我有一个flink集群，其中包含10个任务管理器。我每个人都有一个flink作业在运行。作业也使用广播状态。该广播状态是通过每10分钟读取5个S3文件，进行一些处理，并创建播放的int to list of strings地图来创建的。
问题:在哪里读取文件，是否在JobManager读取和处理文件，并将处理过的内容发送给任务管理人员。
或
是任务管理人员负责所有的读取和处理。如果是这种情况，那么flink如何确保如果任务管理器无法从S3读取，那么所有任务管理器的广播状态都是相同的。
编辑
因此，任务管理器读取
浏览 2提问于2019-10-21得票数 0
我使用Yarn来运行Flink作业。对于每个Flink作业，我都要创建一个检查点。
我提交一个在我的Yarn集群中运行的Flink作业。我有一个轮询作业，它检查一个作业是否在Yarn上失败并重新启动它。当作业再次提交时，Yarn为该Flink作业创建一个新的application_id。如何配置重新提交的Flink作业以使用重新启动的Flink作业的检查点。
我已将conf state.savepoints.dir = hdfs://localhost:9000/checkpoint/设为flink-conf. have‘
创建Flink作业时，streamExecutionEnvironme
浏览 0提问于2018-05-22得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
