前言
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/Shanghai
RUN 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-*.jar
RUN mv ./opt/flink-table-planner_2.12-*.jar ./lib/# 添加用戶自定義代碼（可選）
COPY plugins/ ./plugins/# 設置環境變量
ENV FLINK_HOME=/opt/flink
ENV PATH=$FLINK_HOME/bin:$PATH# 暴露必要的端口
# 8081 - Web UI
# 6123 - TaskManager RPC
EXPOSE 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
# 創建namespace
kubectl create ns bigdata# 創建serviceaccount
kubectl 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: localhostrpc:# 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 <hostname> 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: 6123memory: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: 1600mexecution:# 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: 1memory: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
# # <class-name-of-factory>.
# 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 toaddress: 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:9000
s3.access-key: xxxxxxxx
s3.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資源調度
