flink是一款开源的大数据流式处理框架他可以同时批处理和流处理具有容错性、高吞吐、低延迟等优势本文简述flink在windows和linux中安装步骤和示例程序的运行包括本地调试环境集群环境。另外介绍Flink的开发工程的构建。首先要想运行Flink我们需要下载并解压Flink的二进制包下载地址如下https://flink.apache.org/down…我们可以选择Flink与Scala结合版本这里我们选择最新的1.9版本Apache Flink 1.9.0 for Scala 2.12进行下载。下载成功后在windows系统中可以通过Windows的bat文件或者Cygwin来运行Flink。在linux系统中分为单机集群和Hadoop等多种情况。通过Windows的bat文件运行首先启动cmd命令行窗口进入flink文件夹运行bin目录下的start-cluster.bat注意运行flink需要java环境请确保系统已经配置java环境变量。$cdflink $cdbin $ start-cluster.bat Starting alocalcluster with one JobManager process and one TaskManager process. You can terminate the processes via CTRL-Cinthe spawned shell windows. Web interface by default on http://localhost:8081/.显示启动成功后我们在浏览器访问 http://localhost:8081/可以看到flink的管理页面。通过Cygwin运行Cygwin是一个在windows平台上运行的类UNIX模拟环境官网下载http://cygwin.com/install.html安装成功后启动Cygwin终端运行start-cluster.sh脚本。$cdflink $ bin/start-cluster.sh Starting cluster.显示启动成功后我们在浏览器访问 http://localhost:8081/可以看到flink的管理页面。Linux系统上安装flink单节点安装在Linux上单节点安装方式与cygwin一样下载Apache Flink 1.9.0 for Scala 2.12然后解压后只需要启动start-cluster.sh。集群安装集群安装分为以下几步1、在每台机器上复制解压出来的flink目录。2、选择一个作为master节点然后修改所有机器conf/flink-conf.yamljobmanager.rpc.addressmaster主机名3、修改conf/slaves,将所有work节点写入work01 work024、在master上启动集群bin/start-cluster.sh安装在Hadoop我们可以选择让Flink运行在Yarn集群上。下载Flink for Hadoop的包保证 HADOOP_HOME已经正确设置即可启动 bin/yarn-session.sh运行flink示例程序批处理示例提交flink的批处理examples程序bin/flink run examples/batch/WordCount.jar这是flink提供的examples下的批处理例子程序统计单词个数。$ bin/flink run examples/batch/WordCount.jar Starting execution of program Executing WordCount example with default input data set. Use--inputto specifyfileinput. Printing result to stdout. Use--outputto specify output path.(a,5)(action,1)(after,1)(against,1)(all,2)(and,12)(arms,1)(arrows,1)(awry,1)(ay,1)得到结果这里统计的是默认的数据集可以通过–input --output指定输入输出。我们可以在页面中查看运行的情况流处理示例启动nc服务器nc-l9000提交flink的批处理examples程序bin/flink run examples/streaming/SocketWindowWordCount.jar--port9000这是flink提供的examples下的流处理例子程序接收socket数据传入统计单词个数。在nc端写入单词$nc-l9000lorem ipsum ipsum ipsum ipsum bye输出在日志中$tail-flog/flink-*-taskexecutor-*.out lorem:1bye:1ipsum:4停止flink$ ./bin/stop-cluster.sh在安装好Flink以后只要快速构建Flink工程并完成相关代码开发就可以轻松入手Flink。构建工具Flink项目可以使用不同的构建工具进行构建。为了能够快速入门Flink 为以下构建工具提供了项目模版MavenGradle这些模版可以帮助你搭建项目结构并创建初始构建文件。Maven环境要求唯一的要求是使用Maven 3.0.4或更高版本和安装Java 8.x。创建项目使用以下命令之一来创建项目使用Maven archetypes$ mvn archetype:generate\-DarchetypeGroupIdorg.apache.flink\-DarchetypeArtifactIdflink-quickstart-java\-DarchetypeVersion1.9.0运行quickstart脚本curlhttps://flink.apache.org/q/quickstart.sh|bash-s1.9.0下载完成后查看项目目录结构tree quickstart/ quickstart/ ├── pom.xml └── src └── main ├──java│ └── org │ └── myorg │ └── quickstart │ ├── BatchJob.java │ └── StreamingJob.java └── resources └── log4j.properties示例项目是一个Maven project它包含了两个类StreamingJob和BatchJob分别是DataStreamandDataSet程序的基础骨架程序。main方法是程序的入口既可用于IDE测试/执行也可用于部署。我们建议你将此项目导入IDE来开发和测试它。IntelliJ IDEA 支持 Maven 项目开箱即用。如果你使用的是 Eclipse使用m2e 插件 可以导入 Maven 项目。一些 Eclipse 捆绑包默认包含该插件其他情况需要你手动安装。请注意对 Flink 来说默认的 JVM 堆内存可能太小你应当手动增加堆内存。在 Eclipse 中选择Run Configurations - Arguments并在VM Arguments对应的输入框中写入-Xmx800m。在 IntelliJ IDEA 中推荐从菜单Help | Edit Custom VM Options来修改 JVM 选项。构建项目如果你想要构建/打包你的项目请在项目目录下运行 ‘mvn clean package’ 命令。命令执行后你将找到一个JAR文件里面包含了你的应用程序以及已作为依赖项添加到应用程序的连接器和库target/-.jar。注意如果你使用其他类而不是StreamingJob作为应用程序的主类/入口我们建议你相应地修改pom.xml文件中的mainClass配置。这样Flink 可以从 JAR 文件运行应用程序而无需另外指定主类。Gradle环境要求唯一的要求是使用Gradle 3.x(或更高版本) 和安装Java 8.x。创建项目使用以下命令之一来创建项目Gradle示例build.gradlebuildscript{repositories{jcenter()// this applies only to the GradleShadowplugin}dependencies{classpathcom.github.jengelman.gradle.plugins:shadow:2.0.4}}plugins{idjavaidapplication// shadow plugin to produce fat JARsidcom.github.johnrengelman.shadowversion2.0.4}// artifact properties grouporg.myorg.quickstartversion0.1-SNAPSHOTmainClassNameorg.myorg.quickstart.StreamingJobdescriptionFlink Quickstart Job ext{javaVersion1.8flinkVersion1.9.0scalaBinaryVersion2.11slf4jVersion1.7.7log4jVersion1.2.17}sourceCompatibilityjavaVersion targetCompatibilityjavaVersion tasks.withType(JavaCompile){options.encodingUTF-8}applicationDefaultJvmArgs[-Dlog4j.configurationlog4j.properties]task wrapper(type: Wrapper){gradleVersion3.1}//declarewhere tofindthe dependencies of your project repositories{mavenCentral()maven{urlhttps://repository.apache.org/content/repositories/snapshots/}}// 注意我们不能使用compileOnly或者shadow配置这会使我们无法在 IDE 中或通过使用gradle run命令运行代码。 // 我们也不能从 shadowJar 中排除传递依赖请查看 https://github.com/johnrengelman/shadow/issues/159)。 // -显式定义我们想要包含在flinkShadowJar配置中的类库!configurations{flinkShadowJar // dependencieswhichgo into the shadowJar // 总是排除这些依赖也来自传递依赖因为 Flink 会提供这些依赖。 flinkShadowJar.exclude group:org.apache.flink, module:force-shadingflinkShadowJar.exclude group:com.google.code.findbugs, module:jsr305flinkShadowJar.exclude group:org.slf4jflinkShadowJar.exclude group:log4j}//declarethe dependenciesforyour production andtestcode dependencies{// -------------------------------------------------------------- // 编译时依赖不应该包含在 shadow jar 中 // 这些依赖会在 Flink 的 lib 目录中提供。 // -------------------------------------------------------------- compileorg.apache.flink:flink-java:${flinkVersion}compileorg.apache.flink:flink-streaming-java_${scalaBinaryVersion}:${flinkVersion}// -------------------------------------------------------------- // 应该包含在 shadow jar 中的依赖例如连接器。 // 它们必须在 flinkShadowJar 的配置中 // -------------------------------------------------------------- //flinkShadowJarorg.apache.flink:flink-connector-kafka-0.11_${scalaBinaryVersion}:${flinkVersion}compilelog4j:log4j:${log4jVersion}compileorg.slf4j:slf4j-log4j12:${slf4jVersion}// Addtestdependencies here. // testCompilejunit:junit:4.12}//makecompileOnly dependencies availablefortests: sourceSets{main.compileClasspathconfigurations.flinkShadowJar main.runtimeClasspathconfigurations.flinkShadowJar test.compileClasspathconfigurations.flinkShadowJar test.runtimeClasspathconfigurations.flinkShadowJar javadoc.classpathconfigurations.flinkShadowJar}run.classpathsourceSets.main.runtimeClasspath jar{manifest{attributesBuilt-By:System.getProperty(user.name),Build-Jdk:System.getProperty(java.version)}}shadowJar{configurations[project.configurations.flinkShadowJar]}setting.gradlerootProject.namequickstart或者运行quickstart脚本bash-c$(curlhttps://flink.apache.org/q/gradle-quickstart.sh)--1.9.02.11查看目录结构tree quickstart/ quickstart/ ├── README ├── build.gradle ├── settings.gradle └── src └── main ├──java│ └── org │ └── myorg │ └── quickstart │ ├── BatchJob.java │ └── StreamingJob.java └── resources └── log4j.properties示例项目是一个Gradle 项目它包含了两个类StreamingJob和BatchJob是DataStream和DataSet程序的基础骨架程序。main方法是程序的入口即可用于IDE测试/执行也可用于部署。我们建议你将此项目导入你的 IDE来开发和测试它。IntelliJ IDEA 在安装Gradle插件后支持 Gradle 项目。Eclipse 则通过 Eclipse Buildship 插件支持 Gradle 项目鉴于shadow插件对 Gradle 版本有要求请确保在导入向导的最后一步指定 Gradle 版本 3.0。你也可以使用 Gradle’s IDE integration 从 Gradle 创建项目文件。构建项目如果你想要构建/打包项目请在项目目录下运行 ‘gradle clean shadowJar’ 命令。命令执行后你将找到一个 JAR 文件里面包含了你的应用程序以及已作为依赖项添加到应用程序的连接器和库build/libs/--all.jar。注意如果你使用其他类而不是StreamingJob作为应用程序的主类/入口我们建议你相应地修改build.gradle文件中的mainClassName配置。这样Flink 可以从 JAR 文件运行应用程序而无需另外指定主类。