如何正确使用FlinkStreamSQL

本篇文章为大家展示了如何正确使用FlinkStreamSQL，内容简明扼要并且容易理解，绝对能使你眼前一亮，通过这篇文章的详细介绍希望你能有所收获。

一、前期准备

项目路径：https://github.com/DTStack/flinkStreamSQL

官方文档：https://github.com/DTStack/flinkStreamSQL/blob/1.11_release/docs/quickStart.md

Git Clone 项目

首先，需要将项目从Github上导入到IDEA中（导入方法较多，这里介绍一种常用的）

从IDEA菜单栏里，Git 选项 -> Clone -> FlinkStreamSQL 项目地址，点击Clone即可获取FlinkStreamSQL 源码一份！

项目下载好后，默认分支是 1.11_release，对应的Flink版本是Flink 1.11.x（FlinkStreamSQL 的 release版本对应着 Flink 的 release 版本），需要其他版本的自行切换，推荐使用 1.10_release。

项目编译

项目下载下来后，第一次编译之前，先将整个项目maven reimport 一次

如果有缺少JAR包，在某度或某歌上搜索即可（项目本身并不依赖什么独有的JAR包，毕竟是开源项目），或者在官方钉钉群的文件中搜索看看，会有意外发现。

上面操作没有问题后，就可以开始编译了。

编译命令：

mvn clean package -DskipTests

打包结束后会生成对应的插件包文件夹，1.8 版本对应的是plugins，1.10 及 之后的版本对应的sqlplugins

如果有用不到的插件，可以在项目的root路径下的pom中，注释掉不需要使用的插件

【！！！注意！！！】【！！！注意！！！】【！！！注意！！！】

部分插件之间有依赖关系，所以在注释的时候，请小心别把相关依赖的插件注释掉

rdb模块被所有关系型数据库所依赖，包括impala 模块（虽然它不是关系型数据库，但是它使用了JDBC）

core模块是所有模块所依赖的，不能注释！！

Launcher模块是任务提交必备，不能注释！！

Kafka-base模块是kafka插件的基础，如果使用了kafka插件（不管什么版本），不能注释！！

1.10 及之后的版本，新增了dirtyData模块，是用来提供脏数据指定存储功能（比如将脏数据存储到指定mysql数据库中），不能注释！！

【！！！注意！！！】【！！！注意！！！】【！！！注意！！！】

任务提交

项目编译完之后，就可以提交任务了。任务提交的方式有local、standalone、yarn-session、yarn-per-job模式，后续会支持application（需要等到1.12版本）

从idea提交任务

如果以下概念中，有不懂的，自行查资料了解（学会查资料，比问别人更有效率）

使用的idea版本是2020.3 公开版，有不一样的地方自行修改

这里以yarn-per-job模式为例，其他模式类似，可以看文档自行配置任务提交参数

1.配置idea-application

有个快捷的方法，找到LauncherMain，然后运行，在idea自动生成的application中修改，或者直接"Modify Run Configuration"

这里贴下自己一直使用的任务提交参数，需要的自行修改，每个参数具体什么意思，在官方参数文档中也有详细说明。

-name
Test
-mode
yarnPer
-sql
/dtstack/sql/test/JoinDemoFour.sql
-localSqlPluginPath
/IdeaProjects/StreamSQLOne/sqlplugins
-flinkconf
/dtstack/conf/flink
-yarnconf
/dtstack/conf/yarn
-flinkJarPath
/dtstack/flink-1.10.1/lib
-confProp
{\"metrics.latency.interval\":\"30000\",\"metrics.latency.granularity\":\"operator\",\"time.characteristic\":\"ProcessingTime\",\"disableChain\":\"true\"}
-pluginLoadMode
shipfile
-queue
b

任务SQL怎么写？这个根据自己的插件，去看对应的插件文档，最基本的任务SQL框架是：

CREATE Source（源表） -> CREATE Side（维表，根据自己业务来确定是否需要） -> CREATE Sink（结果表） -> INSERT INTO Sink blablabla...（实际执行的业务SQL，这个必须要，不然任务执行个????）

这里也贴下日常使用的SQL，需要自行修改。

CREATE TABLE SourceOne
(
    id        int,
    name      varchar,
    age       bigint,
    phone     bigint,
    birth     timestamp,
    todayTime time,
    todayDate date,
    money     decimal,
    price     double,
    wechat    varchar,
    proName   varchar
) WITH (
      type = 'kafka11',
      bootstrapServers = 'kudu1:9092',
      zookeeperQuorum = 'kudu1:2181/kafka',
      offsetReset = 'latest',
      topic = 'tiezhu_in_one',
      enableKeyPartitions = 'false',
      topicIsPattern = 'false',
      parallelism = '1'
      );

CREATE TABLE DimOne
(
    id    int,
    age   bigint,
    name  varchar,
    birth timestamp,
    PRIMARY KEY (id, age, name),
    period for system_time
) WITH (
      type = 'mysql',
      url = 'jdbc:mysql://k3:3306/tiezhu?useSSL=false',
      userName = 'root',
      password = 'admin123',
      tableName = 'TestOne',
      parallelism = '1',
      cache = 'LRU',
      asyncCapacity = '100',
      asyncTimeout = '1000',
      errorLimit = '10',
      cacheTTLMs = '1000'
      );

CREATE VIEW ViewOne AS
SELECT DO.age       as age,
       SO.todayTime as todayTime,
       SO.todayDate as todayDate,
       SO.name      as name,
       DO.id        as id,
       DO.birth     as birth,
       SO.proName   as proName
FROM SourceOne SO
         LEFT JOIN DimOne DO
                   ON SO.id = DO.id;

CREATE TABLE DimTwo
(
    id         int,
    proName    varchar,
    createDate date,
    createTime time,
    PRIMARY KEY (id),
    period for system_time
) WITH (
      type = 'mysql',
      url = 'jdbc:mysql://k3:3306/tiezhu?useSSL=false',
      userName = 'root',
      password = 'admin123',
      tableName = 'TestDemoTwo',
      parallelism = '1',
      cache = 'LRU',
      asyncCapacity = '100',
      errorLimit = '10'

      );

CREATE View ViewTwo AS
SELECT DimTwo.proName    as proName,
       DimTwo.createDate as createDate,
       DimTwo.createTime as createTime,
       ViewOne.todayTime as todayTime,
       ViewOne.todayDate as todayDate,
       ViewOne.name      as name,
       ViewOne.birth     as birth,
       ViewOne.age       as age,
       DimTwo.id         as id
FROM ViewOne
         LEFT JOIN DimTwo DimTwo
                   ON ViewOne.id = DimTwo.id
                       and '2020-10-28' = DimTwo.createDate
                       and DimTwo.id >= 2;

CREATE TABLE SinkOne
(
    id         int,
    name       varchar,
    age        bigint,
    birth      timestamp,
    todayTime  time,
    todayDate  date,
    createTime time,
    createDate date,
    proName    varchar
) WITH (
      type = 'kafka11',
      bootstrapServers = 'kudu1:9092',
      topic = 'tiezhu_out',
      parallelism = '1',
      updateMode = 'upsert'
      );

INSERT INTO SinkOne
SELECT *
FROM ViewTwo;

如果需要远程调试，那么需要在flink-conf.yaml中增加Flink 的远程调试配置，然后在idea中配置”JVM Remote“，在代码块中打断点（这种方法还能调试Flink 本身的代码）

env.java.opts.jobmanager: -agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=5005
env.java.opts.taskmanager: -agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=5006

只需要修改标记的这两个地方，如果是HA集群，需要根据日志修改（怎么看日志，怎么修改，自行查资料）

至此，任务远程提交流程就这些。

本地调试

如果嫌弃远程调试，那么可以试试FlinkStreamSQL的本地调试，LocalTest模块（这个模块默认是注释掉的，如果有需要，自行打开即可），使用方法很简单，修改对应的参数，然后执行RUN 即可

但是【注意！！！】LocalTest模块的pom文件中有大部分常用的插件模块，但是如果出现了类似"ClassNotFoundException"，大概率是pom中没有对应的插件模块，同时需要注意，Kafka模块因为有类冲突的存在，所以在LocalTest模块中，Kafka模块只能存在一种

上述内容就是如何正确使用FlinkStreamSQL，你们学到知识或技能了吗？如果还想学到更多技能或者丰富自己的知识储备，欢迎关注亿速云行业资讯频道。