Dinky1.2.3基于Kubernetes Application模式提交Flink作业_dinky history server
前言
Dinky 是一个开箱即用、易扩展,以 Apache Flink 为基础,连接 OLAP 和数据湖等众多框架的一站式实时计算平台,致力于流批一体和湖仓一体的探索与实践。 致力于简化Flink任务开发,提升Flink任务运维能力,降低Flink入门成本,提供一站式的Flink任务开发、运维、监控、报警、调度、数据管理等功能。
今天想给大家说一说,如何通过dinky数据开发平台,将Flink作业采用k8s appliction模式提交到flink集群中。
前置条件
- 需要安装Dinky1.2.3环境
- Flink1.20镜像包
- K8S集群环境
步骤1.构建Flink1.20镜像包
编写Dockerfile文件
FROM flink:1.20.0-scala_2.12-java11# 设置时区(可选)ENV TZ=Asia/ShanghaiRUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone# 设置工作目录WORKDIR /opt/flink# 添加自定义配置文件(可选)# dinky提交作业时并未生效,将会被dinky集群中的flink配置替换COPY conf/config.yaml ./conf/# 添加自定义jar包(可选)COPY lib/*.jar ./lib/# 兼容dinky替换jar包RUN rm -rf ./lib/flink-table-planner-loader-*.jarRUN mv ./opt/flink-table-planner_2.12-*.jar ./lib/# 添加用户自定义代码(可选)COPY plugins/ ./plugins/# 设置环境变量ENV FLINK_HOME=/opt/flinkENV PATH=$FLINK_HOME/bin:$PATH# 暴露必要的端口# 8081 - Web UI# 6123 - TaskManager RPCEXPOSE 8081 6123# 设置容器启动命令(根据需要修改)CMD [\"bash\"]
原始目录结构,lib目录包含了所需jar包,plugins包含插件jia包,conf包含flink配置文件,其中lib中的依赖包,包含了mysql,sqlserver,mongodb,kafka,paimon的依赖,请大家根据实际情况添加依赖包。
依赖jar包下载地址:仓库服务
编译构建完后的镜像完整路径(后续注册集群配置需要用上)
192.168.1.101:5000/bigdata/flink120:latest
步骤2.Kubernetes创建serviceacount
创建命名空间和serviceacount(后续注册集群配置需要用上)
# 创建命名空间和serviceacount# 创建namespacekubectl create ns bigdata# 创建serviceaccountkubectl create serviceaccount flink-service-account -n bigdata# 用户授权kubectl create clusterrolebinding flink-role-binding-flink --clusterrole=edit --serviceaccount=bigdata:flink-service-account# 查看kubectl get pods pods,svc,sa -n bigdata
步骤3.Dinky1.2.3注册集群配置
注册k8s appliction集群配置
################################################################################# Licensed to the Apache Software Foundation (ASF) under one# or more contributor license agreements. See the NOTICE file# distributed with this work for additional information# regarding copyright ownership. The ASF licenses this file# to you under the Apache License, Version 2.0 (the# \"License\"); you may not use this file except in compliance# with the License. You may obtain a copy of the License at## http://www.apache.org/licenses/LICENSE-2.0## Unless required by applicable law or agreed to in writing, software# distributed under the License is distributed on an \"AS IS\" BASIS,# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.# See the License for the specific language governing permissions and# limitations under the License.################################################################################# These parameters are required for Java 17 support.# They can be safely removed when using Java 8/11.env: java: opts: all: --add-exports=java.base/sun.net.util=ALL-UNNAMED --add-exports=java.rmi/sun.rmi.registry=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.file=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.parser=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.tree=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED --add-exports=java.security.jgss/sun.security.krb5=ALL-UNNAMED --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED --add-opens=java.base/java.io=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/java.lang.reflect=ALL-UNNAMED --add-opens=java.base/java.text=ALL-UNNAMED --add-opens=java.base/java.time=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.base/java.util.concurrent=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.locks=ALL-UNNAMED#==============================================================================# Common#==============================================================================jobmanager: # The host interface the JobManager will bind to. By default, this is localhost, and will prevent # the JobManager from communicating outside the machine/container it is running on. # On YARN this setting will be ignored if it is set to \'localhost\', defaulting to 0.0.0.0. # On Kubernetes this setting will be ignored, defaulting to 0.0.0.0. # # To enable this, set the bind-host address to one that has access to an outside facing network # interface, such as 0.0.0.0. bind-host: localhost rpc: # The external address of the host on which the JobManager runs and can be # reached by the TaskManagers and any clients which want to connect. This setting # is only used in Standalone mode and may be overwritten on the JobManager side # by specifying the --host parameter of the bin/jobmanager.sh executable. # In high availability mode, if you use the bin/start-cluster.sh script and setup # the conf/masters file, this will be taken care of automatically. Yarn # automatically configure the host name based on the hostname of the node where the # JobManager runs. address: localhost # The RPC port where the JobManager is reachable. port: 6123 memory: process: # The total process memory size for the JobManager. # Note this accounts for all memory usage within the JobManager process, including JVM metaspace and other overhead. size: 1600m execution: # The failover strategy, i.e., how the job computation recovers from task failures. # Only restart tasks that may have been affected by the task failure, which typically includes # downstream tasks and potentially upstream tasks if their produced data is no longer available for consumption. failover-strategy: regiontaskmanager: # The host interface the TaskManager will bind to. By default, this is localhost, and will prevent # the TaskManager from communicating outside the machine/container it is running on. # On YARN this setting will be ignored if it is set to \'localhost\', defaulting to 0.0.0.0. # On Kubernetes this setting will be ignored, defaulting to 0.0.0.0. # # To enable this, set the bind-host address to one that has access to an outside facing network # interface, such as 0.0.0.0. bind-host: localhost # The address of the host on which the TaskManager runs and can be reached by the JobManager and # other TaskManagers. If not specified, the TaskManager will try different strategies to identify # the address. # # Note this address needs to be reachable by the JobManager and forward traffic to one of # the interfaces the TaskManager is bound to (see \'taskmanager.bind-host\'). # # Note also that unless all TaskManagers are running on the same machine, this address needs to be # configured separately for each TaskManager. host: localhost # The number of task slots that each TaskManager offers. Each slot runs one parallel pipeline. numberOfTaskSlots: 1 memory: process: # The total process memory size for the TaskManager. # # Note this accounts for all memory usage within the TaskManager process, including JVM metaspace and other overhead. # To exclude JVM metaspace and overhead, please, use total Flink memory size instead of \'taskmanager.memory.process.size\'. # It is not recommended to set both \'taskmanager.memory.process.size\' and Flink memory. size: 1728mparallelism: # The parallelism used for programs that did not specify and other parallelism. default: 1# # The default file system scheme and authority.# # By default file paths without scheme are interpreted relative to the local# # root file system \'file:///\'. Use this to override the default and interpret# # relative paths relative to a different file system,# # for example \'hdfs://mynamenode:12345\'# fs:# default-scheme: hdfs://mynamenode:12345#==============================================================================# High Availability#==============================================================================# high-availability:# # The high-availability mode. Possible options are \'NONE\' or \'zookeeper\'.# type: zookeeper# # The path where metadata for master recovery is persisted. While ZooKeeper stores# # the small ground truth for checkpoint and leader election, this location stores# # the larger objects, like persisted dataflow graphs.# ## # Must be a durable file system that is accessible from all nodes# # (like HDFS, S3, Ceph, nfs, ...)# storageDir: hdfs:///flink/ha/# zookeeper:# # The list of ZooKeeper quorum peers that coordinate the high-availability# # setup. This must be a list of the form:# # \"host1:clientPort,host2:clientPort,...\" (default clientPort: 2181)# quorum: localhost:2181# client:# # ACL options are based on https://zookeeper.apache.org/doc/r3.1.2/zookeeperProgrammers.html#sc_BuiltinACLSchemes# # It can be either \"creator\" (ZOO_CREATE_ALL_ACL) or \"open\" (ZOO_OPEN_ACL_UNSAFE)# # The default value is \"open\" and it can be changed to \"creator\" if ZK security is enabled# acl: open#==============================================================================# Fault tolerance and checkpointing#==============================================================================# The backend that will be used to store operator state checkpoints if# checkpointing is enabled. Checkpointing is enabled when execution.checkpointing.interval > 0.# # Execution checkpointing related parameters. Please refer to CheckpointConfig and CheckpointingOptions for more details.# execution:# checkpointing:# interval: 3min# externalized-checkpoint-retention: [DELETE_ON_CANCELLATION, RETAIN_ON_CANCELLATION]# max-concurrent-checkpoints: 1# min-pause: 0# mode: [EXACTLY_ONCE, AT_LEAST_ONCE]# timeout: 10min# tolerable-failed-checkpoints: 0# unaligned: false# state:# backend:# # Supported backends are \'hashmap\', \'rocksdb\', or the# # .# type: hashmap# # Flag to enable/disable incremental checkpoints for backends that# # support incremental checkpoints (like the RocksDB state backend).# incremental: false# checkpoints:# # Directory for checkpoints filesystem, when using any of the default bundled# # state backends.# dir: hdfs://namenode-host:port/flink-checkpoints# savepoints:# # Default target directory for savepoints, optional.# dir: hdfs://namenode-host:port/flink-savepoints#==============================================================================# Rest & web frontend#==============================================================================rest: # The address to which the REST client will connect to address: localhost # The address that the REST & web server binds to # By default, this is localhost, which prevents the REST & web server from # being able to communicate outside of the machine/container it is running on. # # To enable this, set the bind address to one that has access to outside-facing # network interface, such as 0.0.0.0. bind-address: localhost # # The port to which the REST client connects to. If rest.bind-port has # # not been specified, then the server will bind to this port as well. # port: 8081 # # Port range for the REST and web server to bind to. # bind-port: 8080-8090# web:# submit:# # Flag to specify whether job submission is enabled from the web-based# # runtime monitor. Uncomment to disable.# enable: false# cancel:# # Flag to specify whether job cancellation is enabled from the web-based# # runtime monitor. Uncomment to disable.# enable: false#==============================================================================# Advanced#==============================================================================# io:# tmp:# # Override the directories for temporary files. If not specified, the# # system-specific Java temporary directory (java.io.tmpdir property) is taken.# ## # For framework setups on Yarn, Flink will automatically pick up the# # containers\' temp directories without any need for configuration.# ## # Add a delimited list for multiple directories, using the system directory# # delimiter (colon \':\' on unix) or a comma, e.g.:# # /data1/tmp:/data2/tmp:/data3/tmp# ## # Note: Each directory entry is read from and written to by a different I/O# # thread. You can include the same directory multiple times in order to create# # multiple I/O threads against that directory. This is for example relevant for# # high-throughput RAIDs.# dirs: /tmp# classloader:# resolve:# # The classloading resolve order. Possible values are \'child-first\' (Flink\'s default)# # and \'parent-first\' (Java\'s default).# ## # Child first classloading allows users to use different dependency/library# # versions in their application than those in the classpath. Switching back# # to \'parent-first\' may help with debugging dependency issues.# order: child-first# The amount of memory going to the network stack. These numbers usually need# no tuning. Adjusting them may be necessary in case of an \"Insufficient number# of network buffers\" error. The default min is 64MB, the default max is 1GB.## taskmanager:# memory:# network:# fraction: 0.1# min: 64mb# max: 1gb#==============================================================================# Flink Cluster Security Configuration#==============================================================================# Kerberos authentication for various components - Hadoop, ZooKeeper, and connectors -# may be enabled in four steps:# 1. configure the local krb5.conf file# 2. provide Kerberos credentials (either a keytab or a ticket cache w/ kinit)# 3. make the credentials available to various JAAS login contexts# 4. configure the connector to use JAAS/SASL# # The below configure how Kerberos credentials are provided. A keytab will be used instead of# # a ticket cache if the keytab path and principal are set.# security:# kerberos:# login:# use-ticket-cache: true# keytab: /path/to/kerberos/keytab# principal: flink-user# # The configuration below defines which JAAS login contexts# contexts: Client,KafkaClient#==============================================================================# ZK Security Configuration#==============================================================================# zookeeper:# sasl:# # Below configurations are applicable if ZK ensemble is configured for security# ## # Override below configuration to provide custom ZK service name if configured# # zookeeper.sasl.service-name: zookeeper# ## # The configuration below must match one of the values set in \"security.kerberos.login.contexts\"# login-context-name: Client#==============================================================================# HistoryServer#==============================================================================# The HistoryServer is started and stopped via bin/historyserver.sh (start|stop)## jobmanager:# archive:# fs:# # Directory to upload completed jobs to. Add this directory to the list of# # monitored directories of the HistoryServer as well (see below).# dir: hdfs:///completed-jobs/# historyserver:# web:# # The address under which the web-based HistoryServer listens.# address: 0.0.0.0# # The port under which the web-based HistoryServer listens.# port: 8082# archive:# fs:# # Comma separated list of directories to monitor for completed jobs.# dir: hdfs:///completed-jobs/# # Interval in milliseconds for refreshing the monitored directories.# fs.refresh-interval: 10000# # s3密钥配置s3.endpoint: http://192.168.1.102:9000s3.access-key: xxxxxxxxs3.secret-key: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
- 类型:Kubernetes Application
- 集群配置名称:k8s-appliction-test(自定义)
- 暴露端口类型:NodePort
- Kubernetes 命名空间:bigdata (步骤2中创建的命名空间)
- K8s 提交账号:flink-service-account (步骤2中创建的serviceaccount)
- K8s KubeConfig:从k8s服务器中的将文件~/.kube/config内容拷贝过来
- Flink 镜像地址:192.168.1.101:5000/bigdata/flink120:latest (步骤1构建的私有镜像包)
- JobManager CPU 配置:1(根据实际需要配置)
- TaskManager CPU 配置:1(根据实际需要配置)
- Flink 配置文件路径:/usr/local/flink-1.20.0/conf (将flink配置拷贝到dinky服务器中)
- JobManager 内存:1G(根据实际需要配置)
- TaskManager 内存:1G(根据实际需要配置)
- 插槽数:1(根据实际需要配置)
- 保存点路径:s3://flink120/flink-savepoints(统一使用s3作为分布式存储,所以上文中的flink配置新增了s3存储密钥配置)
- 检查点路径:s3://flink120/flink-checkpoints
- Jar 文件路径:s3://flink120/dinky/dinky-app-1.20-1.2.3-jar-with-dependencies.jar
创建作业ods-mysql-to-doris,选择flink集群为k8s-appliction-test
Flink作业运行情况
k8s中自动生成部署Deployments以及pod实例
总结
- 构建私有flink镜像包时,将配置文件拷贝到镜像中,但是dinky提交作业时并未生效,将会被dinky集群中的flink配置替换,实际使用的配置为/usr/local/flink-1.20.0/conf
- flink作业通过k8s appliction模式提交作业,可以抛弃yarn appliction模式,毕竟hadoop框架太重,通过s3(minio/oss)替换了hdfs分布式存储,通过k8s的资源调度替带了yarn资源调度