怎样正确使用FlinkX
这篇文章将为大家详细讲解有关怎样正确使用FlinkX,文章内容质量较高,因此小编分享给大家做个参考,希望大家阅读完这篇文章后对相关知识有一定的了解。
Git Clone 项目
首先将项目从Github中下载并导入到IDEA中(方法较多,介绍常见的一种)
从IDEA菜单栏里,Git 选项 -> Clone -> Flinkx 项目地址,点击Clone即可获取Flinkx 源码一份!
版本对应
项目clone下来默认版本是 1.10_release,目前推荐的稳定版本是 1.8_release 和 1.10_release。最新版本是1.11_release。但是! 最新不代表最稳定。Flinkx的每一个版本,都对应着Flink的版本, 上下版本不一定兼容,所以使用前,请确定好Flinkx与Flinkx对应的版本。
项目编译
因为项目中涉及的插件非常的多,所以编译整个项目耗时非常大,所以建议,对于不需要使用的插件,可以注释掉。同时,强烈建议,先看 官方文档,再去编译
!!!但是,请注意!!!请注意!!!请注意!!!⚠️⚠️⚠️
1. 部分插件之间是有依赖关系的,比如:
flinkx-core是被所有插件依赖,千万不能注释!
flinkx-rdb模块是被所有rdb数据源插件依赖,所以在使用类似mysql,oracle等(包括hive,因为连接时用的是jdbc)的场景下,flinkx-rdb模块千万不能注释!
flinkx-test是用于本地调试代码的模块,项目打包成jar包时是必须要注释掉的!
flinkx-launcher模块是用来将可执行任务提交到对应的执行环境中的,不能注释!!
2. 部分插件中有些jar包在maven中央仓库中可能没有,但是在项目的 jars目录下,开发者贴心地准备了jar。
除此之外,还特意编写了将jar包安装到本地仓库的可执行脚本。 (如何在不同的系统下执行脚本,请自行百度)
[图片截取自 官方文档]
以上内容请注意!!!请注意!!!请注意!!!⚠️⚠️⚠️
项目编译命令
mvn clean package -Dmaven.test.skip
之后在项目的根目录下,生成对应的插件目录
1.8_release -> plugins
1.10_release -> syncplugins
二、任务提交
Flinkx支持的任务提交方式有local-test(通过flinkx-test模块提交任务)、local、standalone、yarn-session、yarn-per-job模式,暂时不支持application模式提交。
从idea中提交任务
使用的 idea版本是 2020.3 公开版,有不一样的地方自行修改
这里以 yarn-per-job模式为例,其他模式类似,可以看 官方文档自行配置任务提交参数
1. 配置idea-application
模块:flinkx-launcher
入口类:Launcher
同样有个快捷的方式,直接在入口类中,在类的执行按钮下,选择"Modify Run Configuration",即可修改
主要修改任务参数,根据自己的情况对应修改即可,下面贴下idea中一直使用的任务参数
-mode yarnPer-name flinkx-test-job docs/example/stream_stream.json-pluginRoot syncplugins-flinkconf /dtstack/conf/flink -yarnconf /dtstack/conf/yarn -flinkLibJar /dtstack/flink-1.10.1/lib -confProp {\"flink.checkpoint.interval"\:60000}-queue c
任务参数具体什么含义?请看官方文档的参数说明
任务JSON怎么配置?请看官方文档,这里就不再多说了,而且,官方还特意整理常用插件的任务JSON,只要对应着修改参数即可,点击链接跳转(如此贴心的官方还有谁?这难道不值得一个star嘛?)
三、调试代码
本地调试
首先打开flinkx-test模块
在LocalTest中 jobPath修改为任务JSON的绝对路径,点击执行即可本地调试。官方还特意添加了常用的参数配置
远程调试
如果需要远程调试,那么需要在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集群,需要根据日志修改(怎么看日志,怎么修改,自行查资料)
关于怎样正确使用FlinkX就分享到这里了,希望以上内容可以对大家有一定的帮助,可以学到更多知识。如果觉得文章不错,可以把它分享出去让更多的人看到。