关于flink:Flink-在有赞的实践和应用

作者：沈磊

简介：明天次要分享的内容是 Flink 在有赞的实际和利用。内容包含：

1. Flink 的容器化革新和实际
2. Flink SQL 的实际和利用
3. 将来布局。

一、Flink 的容器化革新和实际

1. 有赞的集群演进历史

• 2014 年 7 月，第一个 Storm 工作正式上线；
• 2016 年，引入 Spark Streaming，运行在 Hadoop Yarn；
• 2018 年，引入了 Flink，作业模式为 Flink on Yarn Per Job；
• 2020 年 6 月，实现了 100% Flink Jar 工作 K8s 化，K8s 作为 Flink Jar 默认计算资源，Flink SQL 工作 On Yarn，Flink 对立实时开发；
• 2020 年 11 月，Storm 集群正式下线。原先的 storm 工作全副都迁徙到了 Flink；
• 2021 年，咱们打算把所有的 Flink 工作 K8s 化。

2. Flink 在外部反对的业务场景

Flink 反对的业务场景有风控，埋点的实时工作，领取，算法实时特色解决，BI 的实时看板，以及实时监控等等。目前的实时工作规模有 500+。

3. 有赞在 Flink on Yarn 的痛点

次要有三局部：

• 第一，CPU 没有隔离。Flink On Yarn 模式，CPU 没有隔离，某个实时工作造成某台机器 CPU 应用过高时，会对该机器其余实时工作造成影响；
• 第二，大促扩缩容老本高。Yarn 和 HDFS 服务应用物理机，物理机在大促期间扩缩容不灵便，同时须要投入肯定的人力和物力；
• 第三，须要投入人力运维。公司底层利用资源对立为 K8S，独自再对 Yarn 集群运维，会再多一类集群的人力运维老本。

4. Flink on k8s 绝对于 Yarn 的劣势

能够演绎为 4 点：

• 第一，对立运维。公司统一化运维，有专门的部门运维 K8S；
• 第二，CPU 隔离。K8S Pod 之间 CPU 隔离，实时工作不相互影响，更加稳固；
• 第三，存储计算拆散。Flink 计算资源和状态存储拆散，计算资源可能和其余组件资源进行 混部，晋升机器使用率；
• 第四，弹性扩缩容。大促期间可能弹性扩缩容，更好的节俭人力和物力老本。

5. 实时集群的部署状况

总体上分为三层。第一层是存储层；第二层是实时计算资源层；第三层是实时计算引擎层。

• 存储层次要分为两局部：

  • 第一个就是云盘，它次要存储 Flink 工作本地的状态，以及 Flink 工作的日志；
  • 第二局部是实时计算 HDFS 集群，它次要存储 Flink 工作的远端状态。

• 第二层是实时计算的资源层，分为两局部：

  • 一个是 Hadoop Yarn 集群；
  • 另一个是 Flink k8s 集群，再往下细分，会有 Flink k8s 和离线的 HDFS 混部集群的资源，还有 Flink k8s 独自类型的集群资源。
• 最上层有一些实时 Flink Jar，spark streaming 工作，以及 Flink SQL 工作。

咱们思考混部的起因是，离线 HDFS 集群白天机器使用率不高。把离线 HDFS 集群计算资源给实时工作，离线应用外部其余组件的弹性计算资源，从而晋升机器使用率，更好的达到降本成果。

6. Flink on k8s 的容器化流程

如下图所示：

1. 第一步，实时平台的 Flink Jar 工作提交，Flink Jar 工作版本治理，Docker Flink 工作镜像构建，上传镜像到 Docker 镜像仓库；
2. 第二步，工作启动；
3. 第三步，yaml 文件创建；
4. 第四步，和 k8s Api Server 之间进行命令交互；
5. 第五步，从 Docker 镜像仓库拉取 Flink 工作镜像到 Flink k8s 集群；

6. 最初，工作运行。这边有几个 tips：

   • 作业模式为 Flink Standalone Per Job 模式；
   • 每个 Flink Jar 工作一个镜像，通过工作名称 + 工夫截作为镜像的版本；
   • JobManager 须要创立为 Deployment 而不是 Job 类型；
   • Dockerfile 指定 HADOOP_USER_NAME，与线上工作保持一致。

7. 在 Flink on k8s 的一些实际

• 第一个实际是解决资源少配工作无奈启动这个问题。

  先来形容一下问题，Flink on k8s 非云原生，无奈做到实时工作资源按需申请。当用户在平台配置的资源少于实时工作实在应用的资源时（比方用户代码写死并发度，但用户配置的并发度小于该值），会呈现实时工作无奈启动的问题。

  针对这个问题，咱们外部减少了一种 Flink Jar 工作并发度的自动检测机制。它的次要流程如下图所示。首先，用户会在咱们平台去提交 Flink Jar 作业，当他提交实现之后，在后盾会把 Jar 作业以及运行参数，构建 PackagedProgram。通过 PackagedProgram 获取到工作的预执行打算。再通过它获取到工作实在的并发度。如果用户在代码里配置的并发度小于平台端配置的资源，咱们会应用在平台端的配置去申请资源，而后进行启动；反之，咱们会应用它实在的工作并发度去申请资源，启动工作。

• 第二个实际是 Flink on k8s 工作的资源剖析工具。

  首先来说一下背景，Flink k8s 工作资源是用户自行配置，当配置的并发度或者内存过大时，存在计算资源节约的问题，从而会减少底层机器老本。怎么样去解决这个问题，咱们做了一个平台管理员的工具。对于管理员来说，他能够从两种视角去看这个工作的资源是否进行了一个超配：

  • 第一个是工作内存的视角。咱们依据工作的 GC 日志，通过一个开源工具 GC Viewer，拿到这一个实时工作的内存应用指标；
  • 第二个是音讯解决能力的视角。咱们在 Flink 源码层减少了数据源输出 record/s 和工作音讯解决工夫 Metric。依据 metric 找到音讯解决最慢的 task 或者 operator，从而判断并发度配置是否正当。

  管理员依据内存剖析指标以及并发度合理性，联合优化规定，预设置 Flink 资源。而后咱们会和业务方沟通与调整。右图是两种剖析后果，下面是 Flink on K8S pod 内存剖析后果。上面是 Flink K8S 工作解决能力的剖析后果。最终，咱们依据这些指标就能够对工作进行一个资源的从新调整，升高资源节约。目前咱们打算把它做成一个自动化的剖析调整工具。

• 接下来是 Flink on K8s 其余的相干实际。

  • 第一，基于 Ingress Flink Web UI 和 Rest API 的应用。每个工作有一个 Ingress 域名，始终通过域名拜访 Flink Web UI 以及 Resti API 应用；
  • 第二，挂载多个 hostpath volume，解决单块云盘 IO 限度。单块云盘的写入带宽以及 IO 能力有瓶颈，应用多块云盘，升高云盘 Checkpoint 状态和本地写入的压力；
  • 第三，Flink 相干通用配置 ConfigMap 化、Flink 镜像上传胜利的检测。为 Filebeat、Flink 作业通用配置，创立 configmap，而后挂载到实时工作中，确保每个 Flink 工作镜像都胜利上传到镜像仓库；
  • 第四，HDFS 磁盘 SSD 以及基于 Filebeat 日志采集。SSD 磁盘次要是为了升高磁盘的 IO Wait 时 间，调整 dfs.block.invalidate.limit，升高 HDFS Pending delete block 数。工作日志应用 Filebeat 采集，输入到 kafka，前面通过自定义 LogServer 和离线专用 LogServer 查看。

8. Flink on K8s 以后面临的痛点

• 第一，JobManager HA 问题。JobManager Pod 如果挂掉，借助于 k8s Deployment 能力，JobManager 会依据 yaml 文件重启，状态可能会失落。而如果 yaml 配置 Savepoint 复原，则音讯可能大量反复。咱们心愿后续借助于 ZK 或者 etcd 反对 Jobmanager HA；
• 第二，批改代码，再次上传工夫久。一旦代码批改逻辑，Flink Jar 工作上传工夫加上打镜像工夫可能是分钟级别，对实时性要求比拟高的业务或者有影响。咱们心愿后续能够参考社区的实现形式，从 HDFS 下面拉取工作 Jar 运行；
• 第三，K8S Node Down 机，JobManager 复原慢。一旦 K8S Node down 机后，Jobmanager Pod 复原运行须要 8 分钟左右，次要是 k8s 外部异样发现工夫以及作业启动工夫，对局部业务有影响，比方 CPS 实时工作。如何解决，平台端定时检测 K8s node 状态，一旦检测到 down 机状态，将 node 下面有 JobManager 所属的工作进行掉，而后从其之前 checkpoint 复原；
• 第四，Flink on k8s 非云原生。以后通过 Flink Jar 工作并发度自动检测工具解决资源少配无奈启动问题，然而如果工作的预执行打算无奈获取，就无奈获取到代码配置的并发度。咱们的思考是：Flink on k8s 云原生性能以及后面的 1、2 问题，如果社区反对的比拟疾速的话，前面可能会思考将 Flink 版本与社区版本对齐。

9. Flink on K8s 的一些计划举荐

• 第一种计划，是平台本人去构建和治理工作的镜像。

  • 长处是：平台方对于构建镜像，以及运行实时工作整体流程自我掌控，具体问题可能及时修改。
  • 毛病是：须要对 Docker 以及 K8S 相干技术要有肯定理解，门槛应用比拟高，同时须要思考非云原生相干问题。它的实用版本为 Flink 1.6 以上。

• 第二种计划，Flink k8s Operator。

  • 长处是：对用户整体封装了很多底层细节，应用门槛绝对升高一些。
  • 毛病是：整体应用没有第一种计划那么灵便，一旦有问题，因为底层应用的是其封装的性能，底层不好批改。它的实用版本为 Flink 1.7 以上。

• 最初一种计划是，基于社区 Flink K8s 性能。

  • 长处是：云原生，对于资源的申请方面更加敌对。同时，用户应用会更加不便，屏蔽很多底层实现。
  • 毛病是：K8s 云原生性能还是试验中的性能，相干性能还在开发中，比方 k8s Per job 模式。它的实用版本为 Flink 1.10 以上。

二、Flink SQL 实际和利用

1. 有赞 Flink SQL 的倒退历程

• 2019 年 9 月，咱们对 Flink 1.9、1.10 SQL 方面的能力进行钻研和尝试，同时加强了一些 Flink SQL 性能。
• 2019 年 10 月，咱们进行了 SQL 性能验证，基于埋点实时需要，验证 Flink SQL Hbase 维表关联性能，后果合乎预期。
• 2020 年 2 月，咱们对 SQL 的性能进行了扩大，以 Flink 1.10 作为 SQL 计算引擎，进行 Flink SQL 性能扩大开发和优化，实时平台反对全 SQL 化开发。
• 2020 年 4 月，开始反对实时数仓、有赞教育、美业、批发等相干实时需要。
• 2020 年 8 月，新版的实时平台才开始正式上线，目前主推 Flink SQL 开发咱们的实时工作。

2. 在 Flink SQL 方面的一些实际

次要分为三个方面：

• 第一，Flink Connector 的实际包含：Flink SQL 反对 Flink NSQ Connector、Flink SQL 反对 Flink HA Hbase Sink 和维表、Flink SQL 反对无密 Mysql Connector、Flink SQL 反对规范输入（社区曾经反对）、Flink SQL 反对 Clickhouse Sink；
• 第二，平台层的实际包含：Flink SQL 反对 UDF 以及 UDF 治理、反对工作从 Checkpoint 复原、反对幂等函数、反对 Json 相干函数等、反对 Flink 运行相干参数配置，比方状态工夫设置，聚合优化参数等等、Flink 实时工作血统数据自动化采集、Flink 语法正确性检测性能；
• 第三，Flink Runtime 的实际包含：Flink 源码减少单个 Task 以及 Operator 单条记录解决工夫指标；修复 Flink SQL 可撤回流 TOP N 的 BUG。

3. 业务实际

• 第一个实际是咱们外部的客服机器人实时看板。流程分为三层：

  • 第一层是实时数据源，首先是线上的 MySQL 业务表，咱们会把它的 Binlog 通过 DTS 服务同步到相应的 Kafka Topic；
  • 实时工作的 ODS 层有三个 Kafka Topic；

  • 在实时 DWD 层，有两个 Flink SQL 工作。

    • Flink SQL A 生产两个 topic，而后把这两个 topic 外面的数据去通过 Interval Join，依据一些窗口的作用关联到对应的数据。同时，会对这个实时工作设置状态的保留工夫。Join 之后，会去进行一些 ETL 的加工解决，最终会把它的数据输出到一个 topic C。
    • 另外一个实时工作 Flink SQL B 生产一个 topic，而后会对 topic 外面的数据进行荡涤，而后到 HBase 外面去进行一个维表的关联，去关联它所须要的一些额定的数据，关联的数据最终会输出到 topic D。

  在上游，Druid 会生产这两个 topic 的数据，去进行一些指标的查问，最终提供给业务方应用。

• 第二个实际是实时用户行为中间层。用户在咱们平台下面会去搜寻、浏览、退出购物车等等，都会产生相应的事件。原先的计划是基于离线来做的。咱们会把数据落库到 Hive 表，而后算法那边的同学会联合用户特色、机器学习的模型、离线的数据去生成一些用户评分预估，再把它输出到 HBase。

  在这样的背景上面，会有如下诉求：以后的用户评分次要是基于离线工作，而算法同学心愿联合实时的用户特色，更加及时、精确的进步举荐精准度。这其实就须要构建一个实时的用户行为中间层，把用户产生的事件输出到 Kafka 外面，通过 Flink SQL 作业对这些数据进行解决，而后把相应的后果输入到 HBase 外面。算法的同学再联合算法模型，实时的更新模型外面的一些参数，最终实时的进行用户的评分预估，也会落库到 HBase，而后到线上应用。

用户行为中间层的构建流程分为三个步骤：

• 第一层，咱们的数据源在 Kafka 外面；
• 第二层是 ODS 层，在 Flink SQL 作业外面会有一些流表的定义，一些 ETL 逻辑的解决。而后去定义相干的 sink 表、维表等等。这外面也会有一些聚合的操作，而后输出到 Kafka；
• 在 DWS 层，同样有用户的 Flink SQL 作业，会波及到用户本人的 UDF Jar，多流 Join，UDF 的应用。而后去读取 ODS 层的一些数据，落库到 HBase 外面，最终给算法团队应用。

这里有几个实践经验：

• 第一，Kafka Topic、Flink 工作名称，Flink SQL Table 名称，依照数仓命名标准。
• 第二，指标聚合类计算，Flink SQL 工作要设置闲暇状态保留工夫，避免工作状态有限增大。
• 第三，如果存在数据歪斜或者读状态压力较大等状况，须要配置 Flink SQL 优化参数。

4. 在 HAHBase Connector 的实际

社区 HBase Connector 数据关联或者写入是单 HBase 集群应用，当 HBase 集群不可用时，实时工作数据的写入或者关联会受到影响，从而可能会影响到业务应用。至于怎么样去解决这个问题。首先，在 HBase 方面有两个集群，主集群和备集群。它们之间通过 WAL 进行主从的复制。Flink SQL 作业先写入主集群，当主集群不可用的时候，主动降级到备集群，不会影响到线上业务的应用。

5. 无密 Mysql Connector 和指标扩大实际

左图是 Flink 无密 Mysql Sink 语法，解决的问题包含三点：

• 第一，Mysql 数据库用户名和明码不以明文形式向外进行裸露和存储；
• 第二，反对 Mysql 用户名和明码周期性更新；
• 第三，外部主动依据用户名鉴定表权限应用。这样做最次要的目标还是保障实时工作数据库应用更平安。

而后是左下图，咱们在 Flink 源码层面减少 Task 和 Operator 单条音讯解决工夫 Metric。目标是帮忙业务方，依据音讯解决工夫的监控指标，排查和优化 Flink 实时工作。

6. Flink 工作血统元数据自动化采集的实际

Flink 工作血统元数据采集的流程如下图所示，平台启动实时工作后，依据当前任务是 Flink Jar 工作，还是 Flink SQL 工作，别离走两条不同的门路，来获取工作的血统数据，再把血统数据上报元数据系统。这样做的价值有两点：

• 第一，帮忙业务方理解实时工作加工链路。业务方可能更清晰的认知实时工作之间的关系和影响，当操作工作时，可能及时告诉上游其余业务方；
• 第二，更好的构建实时数仓。联合实时工作血统图，提炼实时数据公共层，晋升复用性，更好的构建实时数仓。

三、将来布局

最初是将来的布局，包含四点：

• 第一，推广 Flink 实时工作 SQL 化。推广 Flink SQL 开发实时工作，晋升 Flink SQL 工作比例。
• 第二，Flink 工作计算资源主动优化配置。从内存、工作解决能力、输出速率等，对工作资源进行剖析，对资源配置不合理工作自动化配置，从而升高机器老本。
• 第三，Flink SQL 工作 k8s 化以及 K8s 云原生。Flink 底层计算资源对立为 k8s，升高运维老本，Flink k8s 云原生，更正当应用 K8s 资源。
• 第四，Flink 与数据湖以及 CDC 性能技术的调研。新技术的调研储备，为将来其余实时需要奠定技术根底。