flink是一款開源的大數據流式處理框架，他可以同時批處理和流處理，具有容錯性、高吞吐、低延遲等優勢，本文簡述flink在windows和linux中安裝步驟，和示例程序的運行，包括本地調試環境，集群環境。另外介紹Flink的開發工程的構建。
首先要想運行Flink，我們需要下載並解壓Flink的二進制包，下載地址如下：https://flink.apache.org/downloads.html
我們可以選擇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環境變量。
$ cd flink
$ cd bin
$ start-cluster.bat
Starting a local cluster with one JobManager process and one TaskManager process.
You can terminate the processes via CTRL-C in the 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腳本。
$ cd flink
$ 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.yaml
jobmanager.rpc.address = master主機名
3、修改conf/slaves,將所有work節點寫入
work01
work02
4、在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 --input to specify file input.
Printing result to stdout. Use --output to 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 -l 9000
提交flink的批處理examples程序：
bin/flink run examples/streaming/SocketWindowWordCount.jar --port 9000
這是flink提供的examples下的流處理例子程序，接收socket數據傳入，統計單詞個數。
在nc端寫入單詞
$ nc -l 9000
lorem ipsum
ipsum ipsum ipsum
bye
輸出在日志中
$ tail -f log/flink-*-taskexecutor-*.out
lorem : 1
bye : 1
ipsum : 4
停止flink
$ ./bin/stop-cluster.sh
在安裝好Flink以后，只要快速構建Flink工程，並完成相關代碼開發，就可以輕松入手Flink。
構建工具
Flink項目可以使用不同的構建工具進行構建。為了能夠快速入門，Flink 為以下構建工具提供了項目模版：
這些模版可以幫助你搭建項目結構並創建初始構建文件。
Maven
環境要求
唯一的要求是使用 Maven 3.0.4 （或更高版本）和安裝 Java 8.x。
創建項目
使用以下命令之一來 創建項目：
使用Maven archetypes
$ mvn archetype:generate \
-DarchetypeGroupId=org.apache.flink \
-DarchetypeArtifactId=flink-quickstart-java \
-DarchetypeVersion=1.9.0
運行quickstart腳本
curl https://flink.apache.org/q/quickstart.sh | bash -s 1.9.0
下載完成后，查看項目目錄結構：
tree quickstart/
quickstart/
├── pom.xml
└── src
└── main
├── java
│ └── org
│ └── myorg
│ └── quickstart
│ ├── BatchJob.java
│ └── StreamingJob.java
└── resources
└── log4j.properties
示例項目是一個 Maven project，它包含了兩個類：StreamingJob 和 BatchJob 分別是 DataStream and DataSet 程序的基礎骨架程序。
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.gradle
buildscript {
repositories {
jcenter() // this applies only to the Gradle 'Shadow' plugin
}
dependencies {
classpath 'com.github.jengelman.gradle.plugins:shadow:2.0.4'
}
}
plugins {
id 'java'
id 'application'
// shadow plugin to produce fat JARs
id 'com.github.johnrengelman.shadow' version '2.0.4'
}
// artifact properties
group = 'org.myorg.quickstart'
version = '0.1-SNAPSHOT'
mainClassName = 'org.myorg.quickstart.StreamingJob'
description = """Flink Quickstart Job"""
ext {
javaVersion = '1.8'
flinkVersion = '1.9.0'
scalaBinaryVersion = '2.11'
slf4jVersion = '1.7.7'
log4jVersion = '1.2.17'
}
sourceCompatibility = javaVersion
targetCompatibility = javaVersion
tasks.withType(JavaCompile) {
options.encoding = 'UTF-8'
}
applicationDefaultJvmArgs = ["-Dlog4j.configuration=log4j.properties"]
task wrapper(type: Wrapper) {
gradleVersion = '3.1'
}
// declare where to find the dependencies of your project
repositories {
mavenCentral()
maven { url "https://repository.apache.org/content/repositories/snapshots/" }
}
// 注意：我們不能使用 "compileOnly" 或者 "shadow" 配置，這會使我們無法在 IDE 中或通過使用 "gradle run" 命令運行代碼。
// 我們也不能從 shadowJar 中排除傳遞依賴（請查看 https://github.com/johnrengelman/shadow/issues/159)。
// -> 顯式定義我們想要包含在 "flinkShadowJar" 配置中的類庫!
configurations {
flinkShadowJar // dependencies which go into the shadowJar
// 總是排除這些依賴（也來自傳遞依賴），因為 Flink 會提供這些依賴。
flinkShadowJar.exclude group: 'org.apache.flink', module: 'force-shading'
flinkShadowJar.exclude group: 'com.google.code.findbugs', module: 'jsr305'
flinkShadowJar.exclude group: 'org.slf4j'
flinkShadowJar.exclude group: 'log4j'
}
// declare the dependencies for your production and test code
dependencies {
// --------------------------------------------------------------
// 編譯時依賴不應該包含在 shadow jar 中，
// 這些依賴會在 Flink 的 lib 目錄中提供。
// --------------------------------------------------------------
compile "org.apache.flink:flink-java:${flinkVersion}"
compile "org.apache.flink:flink-streaming-java_${scalaBinaryVersion}:${flinkVersion}"
// --------------------------------------------------------------
// 應該包含在 shadow jar 中的依賴，例如：連接器。
// 它們必須在 flinkShadowJar 的配置中！
// --------------------------------------------------------------
//flinkShadowJar "org.apache.flink:flink-connector-kafka-0.11_${scalaBinaryVersion}:${flinkVersion}"
compile "log4j:log4j:${log4jVersion}"
compile "org.slf4j:slf4j-log4j12:${slf4jVersion}"
// Add test dependencies here.
// testCompile "junit:junit:4.12"
}
// make compileOnly dependencies available for tests:
sourceSets {
main.compileClasspath += configurations.flinkShadowJar
main.runtimeClasspath += configurations.flinkShadowJar
test.compileClasspath += configurations.flinkShadowJar
test.runtimeClasspath += configurations.flinkShadowJar
javadoc.classpath += configurations.flinkShadowJar
}
run.classpath = sourceSets.main.runtimeClasspath
jar {
manifest {
attributes 'Built-By': System.getProperty('user.name'),
'Build-Jdk': System.getProperty('java.version')
}
}
shadowJar {
configurations = [project.configurations.flinkShadowJar]
}
setting.gradle
rootProject.name = 'quickstart'
或者運行quickstart腳本
bash -c "$(curl https://flink.apache.org/q/gradle-quickstart.sh)" -- 1.9.0 2.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 文件運行應用程序，而無需另外指定主類。
Flink系列文章：
Flink入門（一）——Apache Flink介紹
Flink入門（二）——Flink架構介紹
更多實時計算,Flink,Kafka等相關技術博文，歡迎關注實時流式計算
