准备Flink开发环境-使用IntelliJ IDEA+Maven开发Flink项目
摘自拙作《Flink实用教程_1.13.2》一书。
对于熟悉Maven操作的读者来说,可以象上一节讲的那样,先在命令行用Maven命令快速生成Flink基本程序框架,然后导入到IDE中进一步开发。如果读者更熟悉IDE(例如,IntelliJ IDEA)的话,那么可以直接在IntelliJ IDEA中开发Maven程序。通过结合使用IntelliJ IDEA集成开发工具和Maven项目构建工具,我们可以方便快速地开发Flink项目。
一、在IntelliJ IDEA中创建Flink项目
第一步:启动IntelliJ IDEA,创建一个新的项目,如下图所示:
第二步:选择Maven项目,并选择"Create from archetype":
第三步:因为默认没有Flink的archetype,所以需要自己添加。
添加flink-quickstart-java的archetype如下图:
添加flink-quickstart-scala的archetype如下图:
第四步:选择对应的archetype,比如,这里我选择flink-quickstart-scala,如下所示:
第五步:指定项目的groupId、artifactId名称。这里我分别取以下名称:
- groupId:com.xueai8
- artifactId:FlinkScalaDemo
第六步:接下来,指定项目的Maven配置,默认就好。
第七步:指定项目的名称和项目文件所在位置。这里保持默认即可。单击【Finish】按钮,开始创建项目:
第八步:Maven会自动构建项目,最后的项目结构如下所示:
可以看出,flink-quickstart-scala快速地构建了一个基本的Flink项目框架,并生成创建了两个模板程序文件:用于流处理的StreamingJob和用于批处理的BatchJob。
注:同样的步骤,选择flink-quickstart-java,创建一个基于Java API的Flink项目框架。请自行尝试。
二、编写批处理代码并测试执行
以上一节所创建项目中的BatchJob源文件为模板,进一步编写一个简单的批处理代码并执行。这里的目的是为了掌握使用IntelliJ IDEA创建Flink Maven项目,所以不必理解代码,在后续的章节中会详细讲解。
1、在IntelliJ IDEA中打开BatchJob源文件,编辑代码如下:
package com.xueai8
import org.apache.flink.api.scala._
object BatchJob {
def main(args: Array[String]) {
// 设置批执行环境
val env = ExecutionEnvironment.getExecutionEnvironment
// 得到输入数据
val text = env.fromElements("good good study", "day day up")
// 对数据进行转换
val counts = text.flatMap { _.toLowerCase.split("\\W+") }
.map { (_, 1) }
.groupBy(0)
.sum(1)
// 执行并输出结果
counts.print()
}
}
2、在文件内任何空白处,单击右键,在弹出菜单中选择"run BatchJob",执行该程序,在下方的运行窗口可以看到如下输出结果:
(up,1)
(day,2)
(good,2)
(study,1)
注:在执行此程序时,有可能会遇到如下异常:
......
Caused by: java.lang.ClassNotFoundException: org.apache.flink.api.scala.typeutils.CaseClassTypeInfo
at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:335)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
... 13 more
异常的原因是maven引入依赖问题。打开项目中的pom.xml文件,将其中依赖部分的"<scope>provided</scope>"注释(或删除)掉。如下所示:
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-scala_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
<!--<scope>provided</scope>-->
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
<!--<scope>provided</scope>-->
</dependency>
三、项目打包并提交Flink集群执行
项目如果想要部署到生产环境运行,那么就要先打成jar包再进行部署。我们使用Maven的"mvn clean package"命令可以很方便地进行打包。具体操作步骤如下:
1、打开项目中的pom.xml文件,找到以下内容并修改mainClass(可Ctrl + F,查找查找'mainClass')为当前类的全限定名称:
<transformers>
<transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass>com.xueai8.BatchJob</mainClass>
</transformer>
</transformers>
2、在IntelliJ IDEA界面最下方左侧,点击Terminal选项卡,打开Terminal窗口。在打开Terminal窗口窗口中执行命令"mvn clean package"命令打包项目:
3、会编译打包过程中会输出一系列信息,如下所示:
4、在项目下会生成一个target目录,打好的JAR包就在这里,如下图所示:
5、在Linux终端中,执行以下命令,启动Flink集群:
$ cd ~/bigdata/flink-1.10.0/
$ ./bin/start-cluster.sh
过程如下图所示:
6、将该JAR拷贝到Linux的指定目录下,例如,"~/flinkdemos/"目录下。其后提交到Flink集群上运行Job作业:
$ cd ~/bigdata/flink-1.10.0/
$ ./bin/flink run --class com.xueai8.BatchJob ~/flinkdemos/FlinkScalaDemo-1.0-SNAPSHOT.jar
执行结果如下所示:
(day,2)
(good,2)
(study,1)
(up,1)