Apache Flink零基礎(chǔ)入門（二）開發(fā)環(huán)境搭建、應(yīng)用配置部署與運行

在上一篇中，我們介紹了Apache Flink的基本概念。本篇將帶領(lǐng)你從零開始，搭建Flink開發(fā)環(huán)境，并完成一個簡單應(yīng)用的配置、部署與運行全流程。

一、開發(fā)環(huán)境搭建

一個完整的Flink開發(fā)環(huán)境需要以下幾部分：

1. Java開發(fā)環(huán)境：Flink核心基于Java（也支持Scala）。請確保已安裝JDK 8或11（推薦）。
`bash
# 檢查Java版本

java -version
`

2. 構(gòu)建工具：Maven配置
對于Java項目，我們使用Maven管理依賴。在~/.m2/settings.xml中，可以配置鏡像倉庫以加速依賴下載（國內(nèi)用戶建議配置）。
`xml

aliyunmaven
central
阿里云公共倉庫
https://maven.aliyun.com/repository/central

`

集成開發(fā)環(huán)境（IDE）：推薦使用IntelliJ IDEA（社區(qū)版即可）或Eclipse，并安裝好Maven插件。

4. Flink本地安裝（可選，用于本地運行和測試）
從Flink官網(wǎng)下載對應(yīng)版本的二進制包，解壓即可。
`bash
# 解壓后，可以啟動一個本地單節(jié)點集群

./bin/start-cluster.sh
# 訪問Web UI: http://localhost:8081

二、創(chuàng)建第一個Flink應(yīng)用

我們將創(chuàng)建一個Maven項目，實現(xiàn)一個簡單的單詞計數(shù)（WordCount）應(yīng)用。

1. 使用Maven Archetype創(chuàng)建項目
`bash
mvn archetype:generate \

-DarchetypeGroupId=org.apache.flink \

-DarchetypeArtifactId=flink-quickstart-java \

-DarchetypeVersion=1.17.0 \

-DgroupId=com.learn.flink \

-DartifactId=flink-quickstart \

-Dversion=1.0 \

-Dpackage=com.learn.flink \

-DinteractiveMode=false
`

2. 項目核心依賴
查看生成的pom.xml，核心依賴是：
`xml

org.apache.flink
flink-java
${flink.version}

org.apache.flink
flink-streaming-java
${flink.version}

org.apache.flink
flink-clients
${flink.version}

`

3. 編寫WordCount示例代碼
在src/main/java下創(chuàng)建StreamingWordCount.java：
`java
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class StreamingWordCount {
public static void main(String[] args) throws Exception {
// 1. 創(chuàng)建流式執(zhí)行環(huán)境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 2. 定義數(shù)據(jù)源（這里從Socket讀取，用于測試）
DataStreamSource source = env.socketTextStream("localhost", 9999);

// 3. 轉(zhuǎn)換操作：切分、計數(shù)
SingleOutputStreamOperator> result = source
.flatMap(new Tokenizer())
.keyBy(value -> value.f0) // 按單詞分組
.sum(1); // 對第二個字段（計數(shù)）求和

// 4. 輸出結(jié)果（打印到控制臺）
result.print();

// 5. 觸發(fā)程序執(zhí)行（流式作業(yè)必須調(diào)用）
env.execute("Streaming WordCount");
}

// 自定義函數(shù)，將一行文本拆分成（單詞，1）的二元組
public static class Tokenizer implements FlatMapFunction> {
@Override
public void flatMap(String value, Collector> out) {
String[] words = value.toLowerCase().split("\\W+");
for (String word : words) {
if (!word.isEmpty()) {
out.collect(new Tuple2<>(word, 1));
}
}
}
}
}
`

三、應(yīng)用的配置、部署與運行

模式一：本地IDE運行（開發(fā)調(diào)試）

1. 確保有一個Socket源。可以使用nc命令在終端開啟一個服務(wù)：
`bash
# Linux/Mac

nc -lk 9999
# Windows可以使用其他工具，如netcat

在IDE中直接運行StreamingWordCount的main方法。
在nc終端輸入幾行英文句子，即可在IDE控制臺看到實時單詞計數(shù)輸出。

模式二：本地Standalone集群運行

1. 打包應(yīng)用：
`bash
cd flink-quickstart
mvn clean package -DskipTests
`
在target目錄下生成JAR包（如flink-quickstart-1.0.jar）。

2. 提交到本地運行的Flink集群：
`bash
# 首先確保已啟動本地集群（./bin/start-cluster.sh）

./bin/flink run \

-c com.learn.flink.StreamingWordCount \
/path/to/your/flink-quickstart-1.0.jar
`

通過Flink Web UI（http://localhost:8081）監(jiān)控運行中的作業(yè)。
同樣，通過向localhost:9999發(fā)送文本數(shù)據(jù)來觸發(fā)計算。

模式三：提交到生產(chǎn)集群（如YARN/K8s）

以YARN Session模式為例：

1. 啟動YARN Session：
`bash
./bin/yarn-session.sh -tm 2048 -s 2
`

2. 提交作業(yè)：
`bash
./bin/flink run \

-yid \ # 關(guān)聯(lián)到已存在的YARN session

-c com.learn.flink.StreamingWordCount \
/path/to/your/flink-quickstart-1.0.jar
`

四、應(yīng)用配置詳解

Flink應(yīng)用的配置主要通過ExecutionEnvironment或StreamExecutionEnvironment進行。

`java StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 1. 設(shè)置并行度（全局）
env.setParallelism(4);

// 2. 開啟Checkpoint（用于容錯）
env.enableCheckpointing(10000); // 每10秒一次

// 3. 從配置文件讀取配置（如flink-conf.yaml）
// 本地運行時，可加載自定義配置文件
Configuration config = new Configuration();
config.setString("taskmanager.memory.process.size", "2048m");
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
`

還可以通過pom.xml中的<properties>和<profiles>來管理不同環(huán)境（開發(fā)/測試/生產(chǎn)）的構(gòu)建配置。

###

至此，你已經(jīng)完成了從環(huán)境搭建、項目創(chuàng)建、代碼編寫到應(yīng)用部署運行的完整流程。關(guān)鍵步驟是：