草庐IT

Flink模式

yl555 2023-03-28 原文

Per-job Cluster

该模式下,一个作业一个集群,作业之间相互隔离。

在Per-Job模式下,集群管理器框架用于为每个提交的Job启动一个 Flink 集群。Job完成后,集群将关闭,所有残留的资源也将被清除。

此模式可以更好地隔离资源,因为行为异常的Job不会影响任何其他Job。另外,由于每个应用程序都有其自己的JobManager,因此它将记录的负载分散到多个实体中。

场景:Per-Job模式适合长期运行的Job,这些Job可以接受启动延迟的增加以支持弹性。

资源管理器支持:Yarn

Application

与per-job 模式相比,在Application 模式下,main() 方法在集群上而不是在客户端执行。

场景:任务启动较慢,适合于长时间运行的大型任务。

资源管理器支持:Yarn、Native kubernetes

Session

该模式下,作业共享集群资源。Session 模式提交的应用都在该集群里执行,会导致资源的竞争。

该模式优势是无需为每一个提交的任务花费精力去分解集群。但是,如果Job异常或是TaskManager 宕掉,那么该TaskManager运行的其他Job都会失败。除了影响到任务,也意味着潜在需要更多的恢复操作,重启所有的Job,会并发访问文件系统,会导致该文件系统对其他服务不可用。此外,单集群运行多个Job,意味着JobManager更大的负载。

场景:该模式适合于对启动延迟要求较高且运行时间较短的作业,例如交互式查询。任务提交速度快,适合频繁提交运行的短时间任务。

资源管理器支持:Standalone、Yarn、Native kubernetes

Standalone

Standalone模式需要在任务启动时就确定TaskManager的数量,不能像Yarn一样,可以在任务启动时申请动态资源。

很多时候任务需要多少个TaskManager事先并不知道,TaskManager设置少了,任务可能跑不起来,多了又会造成资源浪费,需要在任务启动时才能确定需要多少个TaskMananger。

Standalone Application kubernetes

需要先将用户代码都打到镜像里面,然后根据该镜像来部署一个flink集群运行用户代码。

每提交一个任务,单独启动一个集群运行该任务,运行结束集群被删除,资源也被释放。

Standalone Session kubernetes

在Session模式下,先启动一个Flink集群,然后向该集群提交任务,所有任务共用JobManager。

Native kubernetes

Flink 的 Client 内置了一个 K8s Client,可以借助 K8s Client 去创建 JobManager,当 Job 提交之后,如果对资源有需求,JobManager 会向 Flink 自己的 ResourceManager 去申请资源。这个时候 Flink 的 ResourceManager 会直接跟 K8s 的 API Server 通信,将这些请求资源直接下发给 K8s Cluster,告诉它需要多少个 TaskManger,每个 TaskManager 多大。当任务运行完之后,它也会告诉 K8s Cluster 释放没有使用的资源。相当于 Flink 用很原生的方式了解到 K8s Cluster 的存在,并知晓何时申请资源,何时释放资源。

Native Kubernetes Application

native kubernetes下,application模式相当于提交任务时调k8s api自动拉起一个flink集群跑该应用,然后跑完就删除集群。

这种模式比较适合对启动时间不敏感、且长时间运行的作业。不适合对任务启动时间比较敏感的场景。

优点:隔离性比较好,任务之间资源不冲突,一个任务单独使用一个 Flink 集群;相对于 Flink session 集群而且,资源随用随建,任务执行完成后立刻销毁资源,资源利用率会高一些。

缺点:需要提前指定 TaskManager 的数量,如果 TaskManager 指定的少了会导致作业运行失败,指定的多了仍会降低资源利用率;资源是实时创建的,用户的作业在被运行前,需要先等待以下过程。

flink on native kubernetes application模式:提交任务示例
./bin/flink run-application --target kubernetes-application
-Dkubernetes.namespace=flink-native-kubernetes
-Dkubernetes.cluster-id=flink-application-cluster
-Dkubernetes.jobmanager.service-account=flink
-Dkubernetes.container.image=flink:1.14.2
-Dkubernetes.rest-service.exposed.type=NodePort
-Djobmanager.heap.size=1024m
-Dkubernetes.jobmanager.cpu=1
-Dkubernetes.taskmanager.cpu=2
-Dtaskmanager.memory.process.size=1024m
-Dtaskmanager.numberOfTaskSlots=2
local:///opt/flink/examples/batch/WordCount.jar
应用镜像构建方式:
`FROM flink:1.14.2
 RUN mkdir -p $FLINK_HOME/usrlib
 COPY my-flink-job.jar $FLINK_HOME/usrlib/my-flink-job.jar`

Native Kubernetes Session

native kubernetes下,session模式是提前调k8s api启动一个常驻的flink集群,然后客户端提交任务时,调k8s api自动起一个taskmanager pod 运行任务,然后等任务运行完之后,这个taskmanager的任务pod会被销毁。

flink on native kubernetes session模式:

1、kubectl create namespace flink-session-cluster

2、kubectl create serviceaccount flink -n flink-session-cluster

3、kubectl create clusterrolebinding flink-role-binding-flink --clusterrole=edit --serviceaccount=flink-session-cluster:flink

启动session集群:
4、./bin/kubernetes-session.sh \
 -Dkubernetes.namespace=flink-session-cluster \
-Dkubernetes.jobmanager.service-account=flink \
 -Dkubernetes.cluster-id=flink-session-cluster \
 -Dkubernetes.rest-service.exposed.type=NodePort  \
-Dkubernetes.container.image=flink:1.14.2

flink on native kubernetes session模式:提交任务示例
./bin/flink run -d --target kubernetes-session  \
-Dkubernetes.namespace=flink-session-cluster  \
-Dkubernetes.cluster-id=flink-session-cluster \
 -Dkubernetes.jobmanager.service-account=flink \
-Dkubernetes.rest-service.exposed.type=NodePort \
/opt/flink-1.14.2/examples/batch/WordCount.jar

总结

Flink on K8s :

优点:

Flink 在 K8s 上最简单的方式是以 Standalone 方式进行部署。这种方式部署的好处在于不需要对 Flink 做任何改动,同时 Flink 对 K8s 集群是无感知的,通过外部手段即可让 Flink 运行起来。

缺点:

  • 无论 Operator、Helm Chart 或者是直接使用 Kubectl Yaml 的方式,Flink 都感知不到 K8s 的存在。
  • 目前主要使用静态的资源分配。需要提前确认好需要多少个 TaskManager,如果 Job 的并发需要做一些调整,TaskManager 的资源情况必须相应的跟上,否则任务无法正常执行。
  • 用户需要对一些 Container、Operator 或者 K8s 有一些最基本的认识,这样才能保证顺利将 Flink 运行到 K8s 之上。
  • 对于批处理任务,或者想在一个 Session 里提交多个任务不太友好。无法实时申请资源和释放资源。因为 TaskManager 的资源是固定的,批处理任务可能会分多个阶段去运行,需要去实时地申请资源、释放资源,当前也无法实现。如果需要在一个 Session 里跑多个 Job 并且陆续运行结束当前也无法实现。
  • 如果维持一个比较大的 Session Cluster,可能会资源浪费。但如果维持的 Session Cluster 比较小,可能会导致 Job 跑得慢或者是跑不起来。

基于这几点,社区推进了一个 Native 的集成方案。让Flink 原生的感知到下层 Cluster 的存在。Native 是相对于 Flink 而言的,借助 Flink 的命令就可以达到自治的一个状态,不需要引入外部工具就可以通过 Flink 完成任务在 K8s 上的运行。

生产环境上推荐:

Flink on YARN(pre-job、application)、Flink on Native Kubernetes Appliation;

问题

Flink on Kubernetes 需考虑的问题:

日志问题

日志需要通过k8s的pod日志排查。如果出现节点宕机,pod飘移到别的节点,日志获取困难。

应用jar包问题

flink on k8s的application模式需要将jar包以及依赖放在镜像里启动。

应用依赖问题

有依赖的任务,无法通过客户端获取信息。k8s不支持pre-job模式。

有关Flink模式的更多相关文章

  1. ruby-on-rails - Rails - 子类化模型的设计模式是什么? - 2

    我有一个模型:classItem项目有一个属性“商店”基于存储的值,我希望Item对象对特定方法具有不同的行为。Rails中是否有针对此的通用设计模式?如果方法中没有大的if-else语句,这是如何干净利落地完成的? 最佳答案 通常通过Single-TableInheritance. 关于ruby-on-rails-Rails-子类化模型的设计模式是什么?,我们在StackOverflow上找到一个类似的问题: https://stackoverflow.co

  2. ruby - 解析 RDFa、微数据等的最佳方式是什么,使用统一的模式/词汇(例如 schema.org)存储和显示信息 - 2

    我主要使用Ruby来执行此操作,但到目前为止我的攻击计划如下:使用gemsrdf、rdf-rdfa和rdf-microdata或mida来解析给定任何URI的数据。我认为最好映射到像schema.org这样的统一模式,例如使用这个yaml文件,它试图描述数据词汇表和opengraph到schema.org之间的转换:#SchemaXtoschema.orgconversion#data-vocabularyDV:name:namestreet-address:streetAddressregion:addressRegionlocality:addressLocalityphoto:i

  3. ruby - 如何在续集中重新加载表模式? - 2

    鉴于我有以下迁移:Sequel.migrationdoupdoalter_table:usersdoadd_column:is_admin,:default=>falseend#SequelrunsaDESCRIBEtablestatement,whenthemodelisloaded.#Atthispoint,itdoesnotknowthatusershaveais_adminflag.#Soitfails.@user=User.find(:email=>"admin@fancy-startup.example")@user.is_admin=true@user.save!ende

  4. ruby - 是否有用于序列化和反序列化各种格式的对象层次结构的模式? - 2

    给定一个复杂的对象层次结构,幸运的是它不包含循环引用,我如何实现支持各种格式的序列化?我不是来讨论实际实现的。相反,我正在寻找可能会派上用场的设计模式提示。更准确地说:我正在使用Ruby,我想解析XML和JSON数据以构建复杂的对象层次结构。此外,应该可以将该层次结构序列化为JSON、XML和可能的HTML。我可以为此使用Builder模式吗?在任何提到的情况下,我都有某种结构化数据-无论是在内存中还是文本中-我想用它来构建其他东西。我认为将序列化逻辑与实际业务逻辑分开会很好,这样我以后就可以轻松支持多种XML格式。 最佳答案 我最

  5. ruby-on-rails - environment.rb 中设置的常量在开发模式中消失 - 2

    了解Rails缓存如何工作的人可以真正帮助我。这是嵌套在Rails::Initializer.runblock中的代码:config.after_initializedoSomeClass.const_set'SOME_CONST','SOME_VAL'end现在,如果我运行script/server并发出请求,一切都很好。然而,在我的Rails应用程序的第二个请求中,一切都因单元化常量错误而变得糟糕。在生产模式下,我可以成功发出第二个请求,这意味着常量仍然存在。我已通过将以上内容更改为以下内容来解决问题:config.after_initializedorequire'some_cl

  6. Ruby:标准递归模式 - 2

    我经常迷上ruby​​的一件事是递归模式。例如,假设我有一个数组,它可能包含无限深度的数组作为元素。所以,例如:my_array=[1,[2,3,[4,5,[6,7]]]]我想创建一个方法,可以将数组展平为[1,2,3,4,5,6,7]。我知道.flatten可以完成这项工作,但这个问题是作为我经常遇到的递归问题的一个例子-因此我试图找到一个更可重用的解决方案。简而言之-我猜这种事情有一个标准模式,但我想不出任何特别优雅的东西。任何想法表示赞赏 最佳答案 递归是一种方法,它不依赖于语言。您在编写算法时要考虑两种情况:再次调用函数的情

  7. ruby - 在 Ruby 中查找多个正则表达式匹配的模式和位置 - 2

    这应该是一个简单的问题,但我找不到任何相关信息。给定一个Ruby中的正则表达式,对于每个匹配项,我需要检索匹配的模式$1、$2,但我还需要匹配位置。我知道=~运算符为我提供了第一个匹配项的位置,而string.scan(/regex/)为我提供了所有匹配模式。如果可能,我需要在同一步骤中获得两个结果。 最佳答案 MatchDatastring.scan(regex)do$1#Patternatfirstposition$2#Patternatsecondposition$~.offset(1)#Startingandendingpo

  8. ruby - sinatra 框架的 MVC 模式 - 2

    我想开始使用“Sinatra”框架进行编码,但我找不到该框架的“MVC”模式。是“MVC-Sinatra”模式或框架吗? 最佳答案 您可能想查看Padrino这是一个围绕Sinatra构建的框架,可为您的项目提供更“类似Rails”的感觉,但没有那么多隐藏的魔法。这是使用Sinatra可以做什么的一个很好的例子。虽然如果您需要开始使用这很好,但我个人建议您将它用作学习工具,以对您来说最有意义的方式使用Sinatra构建您自己的应用程序。写一些测试/期望,写一些代码,通过测试-重复:)至于ORM,你还应该结帐Sequel其中(imho

  9. ruby-on-rails - Rails 如何创建数据模式种子数据 - 2

    有没有一种方法可以自动生成种子数据文件并创建种子数据,就像您在下面链接中的Laravel中看到的那样?LaravelDatabaseMigrations&Seed我在另一个应用程序上看到在Rails的db文件夹下创建了一些带有时间戳的文件,其中包含种子数据。创建它的好方法是什么? 最佳答案 我建议你使用Fabrication的组合gem和Faker.Fabrication允许您编写一个模式来构建您的对象,而Faker为您提供虚假数据,如姓名、电子邮件、电话号码等。这是制造商的样子:Fabricator(:user)dousernam

  10. ruby-on-rails - Ruby on Rails 应用程序的只读模式 - 2

    我有一个交互式RubyonRails应用程序,我想在特定时间将其置于“只读模式”。这将允许用户读取他们需要的数据,但阻止他们执行写入数据库的操作。执行此操作的一种方法是在数据库中放置一个true/false变量,该变量在进行任何写入之前进行检查。我的问题。有没有更优雅的解决方案来解决这个问题? 最佳答案 如果你真的想阻止任何数据库写入,我能想到的最简单的方法是覆盖readonly?始终返回true的模型方法,无论是在选定模型中还是对于所有ActiveRecord模型。如果模型设置为只读(通常通过调用#readonly!来完成),任何

随机推荐