一、前期准备
项目路径：https://github.com/DTStack/flinkStreamSQL
官方文档：https://github.com/DTStack/flinkStreamSQL/blob/1.11_release/docs/quickStart.md
官方视频：b站（欢迎各位多多三连）
https://www.bilibili.com/video/BV1GA411J7RL?from=search&seid=9301258683583870508 《FlinkStreamSQL初步介绍》
https://www.bilibili.com/video/BV14f4y1D7yR?from=search&seid=9301258683583870508 《FlinkStreamSQL的使用和贡献》
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模块只能存在一种
![]()
