flinkstream flinksql 优缺点(flink sql和datastream的区别)

本文向您展示了如何正确使用FlinkStreamSQL。内容简洁易懂。一定会让你眼前一亮。希望通过这篇文章的详细介绍，你能有所收获。

一、前期准备

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

正式文件：https://github.com/DTStack/FlinkStreamsQL/blob/1.11发布/文档/快速启动

00-1010首先，我们需要将项目从Github导入IDEA(导入方式有很多，这里是常用的一种)

从IDEA菜单栏，Git选项-Clone-FlinkStreamSQL项目地址，点击Clone获取FlinkStreamSQL源代码的副本！

项目下载后，默认分支是1.11_release,对应的Flink版本是Flink 1.11.x(FlinkStreams SQL的发布版本对应FlinkStreams SQL的发布版本)。其他版本需要自行切换，推荐1.10_release。

Git Clone 项目

项目下载后，第一次编译前，maven将整个项目重新导入一次。

如果有缺失的JAR包，只需在某个度或歌曲上搜索(项目本身并不依赖任何唯一的JAR包，毕竟是开源项目)，或者在官方钉子户群的文件中搜索，就会发现一些意想不到的东西。

当上述操作没有问题时，就可以开始编译了。

编译命令：

相应的插件包文件夹将在打包mvncleanpackage-DskipTests后生成。1.8版对应插件，1.10版及更高版本对应sqlplugins。

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

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

有些插件之间是有依赖关系的，所以评论的时候请注意不要注释掉依赖的插件。

包括rdb在内的所有关系型数据库,依赖impala 模块模块(虽然它不是关系数据库，但它使用JDBC)

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

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

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

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

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

项目编译

项目编制完成后，即可提交任务。任务提交有本地、独立、纱线会话、每任务码模式，稍后将支持应用程序(需要等到1.12版本)。

00-1010如果您不理解以下任何概念，请自行检查信息(学会查资料，比问别人更有效率)

使用的idea版本是2020.3 公开版,有不同的地方可以修改。

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

1.配置想法-应用程序

linkStreamSQL">

有个快捷的方法，找到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，你们学到知识或技能了吗？如果还想学到更多技能或者丰富自己的知识储备，欢迎关注行业资讯频道。