使用IntelliJ IDEA+Maven开发Flink项目

您所在的位置:网站首页 maven创建项目后不能创建Java文件 使用IntelliJ IDEA+Maven开发Flink项目

使用IntelliJ IDEA+Maven开发Flink项目

2023-10-11 23:50| 来源: 网络整理| 查看: 265

对于熟悉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.xueai8artifactId: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源文件,编辑代码如下:

public static void main(String[] args) throws Exception { // set up the batch execution environment final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); DataSource stringDataSource = env.fromElements("ssssss yi ss", "ss jiu sss"); DataSet counts = // 把每一行文本切割成二元组,每个二元组为: (word,1) stringDataSource.flatMap(new Tokenizer()) // 根据二元组的第“0”位分组,然后对第“1”位求和 .groupBy(0) .sum(1); counts.print(); } // 自定义函数 public static class Tokenizer implements FlatMapFunction { @Override public void flatMap(String value, Collector out) { // 统一大小写并把每一行切割为单词 String[] tokens = value.toLowerCase().split("\\W+"); // 消费二元组 for (String token : tokens) { if (token.length() > 0) { out.collect(new Tuple2(token, 1)); } } } }

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文件,将其中依赖部分的"provided"注释(或删除)掉。如下所示:

org.apache.flink flink-scala_${scala.binary.version} ${flink.version} org.apache.flink flink-streaming-scala_${scala.binary.version} ${flink.version} 三、项目打包并提交Flink集群执行

项目如果想要部署到生产环境运行,那么就要先打成jar包再进行部署。我们使用Maven的"mvn clean package"命令可以很方便地进行打包。具体操作步骤如下:

1、打开项目中的pom.xml文件,找到以下内容并修改mainClass(可Ctrl + F,查找查找'mainClass')为当前类的全限定名称:

com.xueai8.BatchJob

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

执行结果如下所示:

对自定义的Maven Archetype管理,找到C:\Users\95844\.IntelliJIdea2019.2\system\Maven\Indices\UserArchetypes.xml进行编辑删除即可。

(day,2) (good,2) (study,1) (up,1)


【本文地址】


今日新闻


推荐新闻


    CopyRight 2018-2019 办公设备维修网 版权所有 豫ICP备15022753号-3