Flink 与 Flink 可视化平台 StreamPark 教程（开篇）

本文分享自天翼云开发者社区《Flink 与Flink可视化平台StreamPark教程（开篇）》，作者：l****n

介绍

Flink 是一个大数据流处理引擎，可以为不同行业提供实时大数据处理解决方案。随着 Flink 的快速发展和改进，世界各地的许多公司现在都能看到它的存在。目前，北美、欧洲和金砖国家都是全球 Flink 应用的热门地区。当然，Flink 在中国的知名度特别高，部分原因是一些互联网大厂的贡献和引领效应，也符合中国的反应与场景密切相关。想象一下，在中国，一个网站可能需要面对数以亿计的日活跃用户和每秒数亿的计算峰值，这对许多外国公司来说是难以想象的。Flink 为我们提供了高速准确处理海量流媒体数据的可能性。

在目前的云原生时代，容器化、K8S 等技术已经在各个互联网大厂中独占鳌头，大部分的应用已经实现了上云。对于大数据引擎家族中的一员，flink 实现与 K8S 结合、实现云原生下的 severless 模式的需求日渐增加，。因此，在本文中，主要为实现面对云原生+flink 进行讲解，希望能够给读者带来获得新知识的喜悦。

在这里，将会提供 flink 的使用方法，和一个 flink 可视化平台 StreamPark 中的使用方式。本文将实时更新，将依次介绍其中各个方式的使用方法。在这里将会涉及以下知识点：

• DataStreamApi 的使用

• UDF 的开发

• FlinkSql 的使用

• Flink cdc 功能

• 原生 flink k8s application 的使用

• 翼 flink-StreamPark 的使用要点

本文的目录暂定如此，后续将会对其中的内容加以补充，请广大读者提出宝贵意见，如需添加或删减某些知识点可留言或私信本文作者。

基础环境

在本文中，将面向开发程序员、面向一线码农，带来最详细的 flink 教程。从基础环境搭建到最后的平台应用均会涉及。

对于 flink 而言，少不了对流式数据的处理，一般而言面对 kafka、rabbitmq、cdc 等消息为数据源主流，在这里，为简化基础环境搭建流程，将提供 mysql 数据源并开启 binlog 模式作为我们的数据源，实现流（CDC 功能接入 binlog）批（常规查询）一体的输入。

数据源搭建

在本文中，我们使用 mysql 作为数据源，并开启 binlog 作为流数据作为本实例中的数据源。在这里首先需要安装一个 docker 运行 mysql 容器，已实现统一基础环境。

# 移除掉旧的版本sudo yum remove docker \                docker-client \                docker-client-latest \                docker-common \                docker-latest \                docker-latest-logrotate \                docker-logrotate \                docker-selinux \                docker-engine-selinux \                docker-engine​# 删除所有旧的数据sudo rm -rf /var/lib/docker​# 安装依赖包sudo yum install -y yum-utils \device-mapper-persistent-data \lvm2​# 添加源，使用了阿里云镜像sudo yum-config-manager \   --add-repo \  http://mirrors.aliyun.com/docker-ce/linux/centos/docker-ce.repo​# 配置缓存sudo yum makecache fast​# 安装最新稳定版本的dockersudo yum install -y docker-ce​# 配置镜像加速器sudo mkdir -p /etc/dockersudo tee /etc/docker/daemon.json <<-'EOF'{ "registry-mirrors": ["http://hub-mirror.c.163.com"]}EOF​# 启动docker引擎并设置开机启动sudo systemctl start dockersudo systemctl enable docker​# 配置当前用户对docker的执行权限sudo groupadd dockersudo gpasswd -a ${USER} dockersudo systemctl restart docker

完成 docker 的安装后，可以执行如下命令，实现 mysql 的安装

docker run -p 3307:3306 --name myMysql -v /mydata/mysql/log:/var/log/mysql -v /mydata/mysql/data:/var/lib/mysql -v /mydata/mysql/conf:/etc/mysql -e MYSQL_ROOT_PASSWORD=*****  -d mysql:5.7.25

注意这里我们建议开启 mysql 的 binlog 功能，供我们后续的 CDC 功能的使用，因此在启动后需修改 mysql 的配置文件，以使其支持 binlog 功能。开启此功能后，关于 mysql 中数据的修改将会被记录，在后续连接 mysql 后，将会以流

修改 my.cnf 文件

[mysqld]log-bin=/var/lib/mysql/mysql-binserver-id=123654expire_logs_days = 30

之后重启容器

docker restart myMysql

构键 k8s 集群

1. 在这里，我们需要搭建一个 K8S 环境用于提供 flink 任务的运行时环境。在这里推荐使用 kubeadm 或者一些脚本工具【链接】github中的脚本工具搭建。具体过程在这里省略，可以参考上述链接中的文档进行操作。

2. 需要注意的是，我们需要在相应用户的目录下提供一个 kubeconfig 文件，一般而言，该文件在安装好 k8s 后将会在~/.kube/目录下出现，如下图所示，通过该文件，才能顺利地调用 K8S 客户端提交任务，该 config 的内容为与 K8S 的 ApiServer 进行连接时需要使用的信息。

    

下载 flink 客户端

flink 客户端是控制 flink 的核心，需要下载并部署

wget https://archive.apache.org/dist/flink/flink-1.14.3/flink-1.14.3-bin-scala_2.12.tgztar -xf flink-1.14.3-bin-scala_2.12.tgz

提供 flink 运行任务的环境

1. 将 kubeconfig 提供出来，供 flink 客户端调用，在这里要保证我们使用的客户端时，我们的用户下拥有 kubeconfig 文件

2. 在这里主要提供一个供 flink 使用的命名空间、和 SA。在 K8S Application 模式下，service acount（SA）是 flink 的 jobmanager 使用的服务账号，jobmanager 以此来获得启动相应的 taskamanager 的权限。这一点在后续的 K8S application 模式下比较重要。

# 创建namespacekubectl create ns flink-dev# 创建serviceaccountkubectl create serviceaccount flink-service-account -n flink-dev# 用户授权kubectl create clusterrolebinding flink-role-binding-flink --clusterrole=edit --serviceaccount=flink-dev:flink-service-accoun