（6）Flink SQL on k8s 实现

1.简介

我们在实际使用flink的过程中，不仅使用java开发flink作业，同时也会直接使用flink SQL，通过编写SQL的方式，来实现flink作业。本文就基于Flink 官方提供的 Kubernetes Operator，来实现flink sql在k8s上的运行。

2.程序功能示意图

开发一个通用的flink sql运行程序，通过向程序的启动类传参的方式，将sql脚本的路径传给程序，然后程序就会读取sql脚本，解析出里面的sql语句，并把他们转换为flink 任务提交

sql脚本的逻辑主要是从Kafka读取数据，并经过flink处理，将数据写入到mysql

在我们这个案例中，我们只需要开发好SqlRunner，具体的逻辑都在sql脚本里面定义，这样只需要编写不同的脚本就可以，让程序变得非常通用

3.Flink SQL程序开发

主程序：

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.FileUtils;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.util.ArrayList;
import java.util.List;

public class SqlRunner {

    private static final Logger LOG = LoggerFactory.getLogger(SqlRunner.class);

    private static final String STATEMENT_DELIMITER = ";"; // a statement should end with `;`
    private static final String LINE_DELIMITER = "n";

    private static final String COMMENT_PATTERN = "(--.*)|(((\/\*)+?[\w\W]+?(\*\/)+))";

    public static void main(String[] args) throws Exception {
        String scriptFilePath = "simple.sql";
        if (args.length == 1) {
            scriptFilePath = args[0];
        }

        String script = FileUtils.readFileUtf8(new File(scriptFilePath));
        List statements = parseStatements(script);

        //建立Stream环境，设置并行度为1
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        //建立Table环境
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);
        TableResult tableResult = null;
        for (String statement : statements) {
            LOG.info("Executing:n{}", statement);
            System.out.println("Executing: " + statement);
            tableResult = tableEnvironment.executeSql(statement);
        }
//        ** executeSql是一个异步的接口，在idea里面跑的话，直接就结束了，需要手动拿到那个executeSql的返回的TableResult
        tableResult.getJobClient().get().getJobExecutionResult().get();
    }

    public static List parseStatements(String script) {
        String formatted = formatSqlFile(script).replaceAll(COMMENT_PATTERN, "");
        List statements = new ArrayList();

        StringBuilder current = null;
        boolean statementSet = false;
        for (String line : formatted.split("n")) {
            String trimmed = line.trim();
            if (trimmed == null || trimmed.length() < 1) {
                continue;
            }
            if (current == null) {
                current = new StringBuilder();
            }
            if (trimmed.startsWith("EXECUTE STATEMENT SET")) {
                statementSet = true;
            }
            current.append(trimmed);
            current.append("n");
            if (trimmed.endsWith(STATEMENT_DELIMITER)) {
                if (!statementSet || trimmed.equals("END;")) {
                    // SQL语句不能以分号结尾
                    statements.add(current.toString().replace(";", ""));
                    current = null;
                    statementSet = false;
                }
            }
        }
        return statements;
    }

    public static String formatSqlFile(String content) {
        String trimmed = content.trim();
        StringBuilder formatted = new StringBuilder();
        formatted.append(trimmed);
        if (!trimmed.endsWith(STATEMENT_DELIMITER)) {
            formatted.append(STATEMENT_DELIMITER);
        }
        formatted.append(LINE_DELIMITER);
        return formatted.toString();
    }
}

sql 脚本文件：放在程序的根目录下

-- source到kafka
CREATE TABLE user_log
(
    user_id     VARCHAR,
    item_id     VARCHAR,
    category_id VARCHAR,
    behavior    VARCHAR,
    ts          TIMESTAMP
) WITH (
      'connector' = 'kafka',
      'topic' = 'user_behavior',
      'properties.bootstrap.servers' = '172.16.252.129:9092,172.16.252.130:9092,172.16.252.131:9092',
      'properties.group.id' = 'testGroup1',
      'scan.startup.mode' = 'earliest-offset',
      'format' = 'json'
      );

-- sink到mysql
CREATE TABLE pvuv_sink (
                           dt VARCHAR,
                           pv BIGINT,
                           uv BIGINT,
                           primary key (dt) not enforced
) WITH (
      'connector' = 'jdbc',
      'url' = 'jdbc:mysql://172.16.252.113:3306/flink_test?characterEncoding=utf8&connectTimeout=1000&socketTimeout=3000&autoReconnect=true&useUnicode=true&useSSL=false&serverTimezone=UTC',
      'table-name' = 'pvuv_sink',
      'username' = 'root',
      'password' = '111...aaa'
      );

-- 打印到屏幕
-- CREATE TABLE pvuv_sink
-- (
--     dt VARCHAR,
--     pv BIGINT,
--     uv BIGINT
-- ) WITH (
--       'connector' = 'print'
--       );

-- 把统计结果数据保存到MySQL
INSERT INTO pvuv_sink
SELECT DATE_FORMAT(ts, 'yyyy-MM-dd HH:00') dt,
       COUNT(*)                AS          pv,
       COUNT(DISTINCT user_id) AS          uv
FROM user_log
GROUP BY DATE_FORMAT(ts, 'yyyy-MM-dd HH:00');

mysql 建表语句：

CREATE TABLE pvuv_sink (
    dt VARCHAR(30),
    pv BIGINT,
    uv BIGINT,
    primary key (dt)
)

本地运行，往Kafka 发送消息，准备的数据是json格式的，要和Kafka定义表的数据格式一致

{"user_id": "543462", "item_id":"1715", "category_id": "1464116", "behavior": "pv", "ts": "2023-03-26 01:00:00"}
{"user_id": "543462", "item_id":"1715", "category_id": "1464116", "behavior": "pv", "ts": "2023-03-26 02:00:00"}
{"user_id": "662867", "item_id":"2244074", "category_id": "1575622", "behavior": "pv", "ts": "2023-03-26 03:36:00"}
{"user_id": "662867", "item_id":"2244074", "category_id": "1575622", "behavior": "pv", "ts": "2023-03-26 04:36:00"}

测试通过后，打包把相关的依赖范围改为provided，减少jar包体积。

    org.apache.flink
    flink-connector-kafka_${scala.binary.version}
    ${flink.version}
    provided


    org.apache.flink
    flink-connector-jdbc_${scala.binary.version}
    ${flink.version}
    provided


    mysql
    mysql-connector-java
    5.1.49
    provided




    org.apache.flink
    flink-connector-files
    ${flink.version}
    provided


    org.apache.flink
    flink-table-api-java
    ${flink.version}
    provided


    org.apache.flink
    flink-table-api-java-bridge_${scala.binary.version}
    ${flink.version}
    provided


    org.apache.flink
    flink-table-planner-blink_${scala.binary.version}
    ${flink.version}
    provided


    org.apache.flink
    flink-csv
    ${flink.version}
    provided


    org.apache.flink
    flink-json
    ${flink.version}
    provided

上传jar包和simple.sql到flink-jar-pvc对应的pv目录下

4.Flink SQL镜像构建

因为JobManager启动的时候就要去连接kafka和mysql，而flink自有lib目录下没有kafka和mysql的jar包，所以需要新构建flink-sql的镜像，将kafka和mysql的jar包放到/opt/flint/lib目录下

mkdir flink-sql
cd flink-sql/
wget https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-kafka_2.12/1.13.6/flink-sql-connector-kafka_2.12-1.13.6.jar
wget https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-jdbc_2.12/1.13.6/flink-connector-jdbc_2.12-1.13.6.jar
wget https://repo.maven.apache.org/maven2/mysql/mysql-connector-java/5.1.49/mysql-connector-java-5.1.49.jar

编写Dockerfile

[root@k8s-demo001 flink-sql]# cat Dockerfile 
FROM flink:1.13.6
WORKDIR /opt/flink
COPY flink-connector-jdbc_2.12-1.13.6.jar /opt/flink/lib/flink-connector-jdbc_2.12-1.13.6.jar
COPY flink-sql-connector-kafka_2.12-1.13.6.jar /opt/flink/lib/flink-sql-connector-kafka_2.12-1.13.6.jar
COPY mysql-connector-java-5.1.49.jar /opt/flink/lib/mysql-connector-java-5.1.49.jar
RUN chown -R flink:flink /opt/flink/lib
ENTRYPOINT ["/docker-entrypoint.sh"]
EXPOSE 6123 8081
CMD ["help"]

构建镜像，并上传到镜像仓库

docker build -f Dockerfile -t flink-sql:1.13.6 .
docker tag flink-sql:1.13.6 172.16.252.110:8011/flinkonk8s/flink-sql:1.13.6
docker push 172.16.252.110:8011/flinkonk8s/flink-sql:1.13.6

5.提交flink-sql作业

1、编写flink-sql作业yaml并提交

[root@k8s-demo001 ~]# cat sql-application.yaml 
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  namespace: flink
  name: sql-application
spec:
  image: 172.16.252.110:8011/flinkonk8s/flink-sql:1.13.6  # 使用flink-sql的镜像
  flinkVersion: v1_13
  imagePullPolicy: IfNotPresent  # 镜像拉取策略，本地没有则从仓库拉取
  ingress:   # ingress配置，用于访问flink web页面
    template: "flink.k8s.io/{{namespace}}/{{name}}(/|$)(.*)"
    className: "nginx"
    annotations:
      nginx.ingress.kubernetes.io/rewrite-target: "/$2"
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "2"
    classloader.resolve-order: parent-first  # 默认是child-first，必须改为parent-first，先加载flink自带的Jar，要不然容易出现各种类型不匹配问题
  serviceAccount: flink
  jobManager:
    resource:
      memory: "1024m"
      cpu: 1
  taskManager:
    resource:
      memory: "1024m"
      cpu: 1
  podTemplate:
    spec:
      hostAliases:
        - ip: "172.16.252.129"
          hostnames:
            - "Kafka-01"
        - ip: "172.16.252.130"
          hostnames:
            - "Kafka-02"
        - ip: "172.16.252.131"
          hostnames:
            - "Kafka-03"
      containers:
        - name: flink-main-container
          env:
            - name: TZ
              value: Asia/Shanghai
          volumeMounts:
            - name: flink-jar  # 挂载nfs上的jar
              mountPath: /opt/flink/jar
            - name: flink-log
              mountPath: /opt/flink/log
      volumes:
        - name: flink-jar
          persistentVolumeClaim:
            claimName: flink-jar-pvc
        - name: flink-log
          persistentVolumeClaim:
            claimName: flink-log-pvc
  job:
    jarURI: local:///opt/flink/jar/flink-on-k8s-demo-1.0-SNAPSHOT-jar-with-dependencies.jar # 使用pv方式挂载jar包
    entryClass: org.fblinux.SqlRunner
    args:   # 传递到作业main方法的参数
      - "/opt/flink/jar/simple.sql"
    parallelism: 1
    upgradeMode: stateless
[root@k8s-demo001 ~]# kubectl apply -f sql-application.yaml

 验证pod创建情况

通过Kafka，发送测试数据

{"user_id": "543462", "item_id":"1715", "category_id": "1464116", "behavior": "pv", "ts": "2023-03-26 11:00:00"}
{"user_id": "543462", "item_id":"1715", "category_id": "1464116", "behavior": "pv", "ts": "2023-03-26 12:00:00"}
{"user_id": "662867", "item_id":"2244074", "category_id": "1575622", "behavior": "pv", "ts": "2023-03-26 13:36:00"}
{"user_id": "662867", "item_id":"2244074", "category_id": "1575622", "behavior": "pv", "ts": "2023-03-26 14:36:00"}