# 1. 下载Flink
cd /bigdata/app
wget https://archive.apache.org/dist/flink/flink-1.17.1/flink-1.17.1-bin-scala_2.12.tgz
tar -zxvf flink-1.17.1-bin-scala_2.12.tgz
ln -s flink-1.17.1 flink

# 2. 配置flink-conf.yaml
cat > /bigdata/app/flink/conf/flink-conf.yaml << ‘EOF’
jobmanager.rpc.address: fgedu-jm01
jobmanager.memory.process.size: 8g
taskmanager.memory.process.size: 32g
taskmanager.numberOfTaskSlots: 4
parallelism.default: 4

state.backend: rocksdb
state.checkpoints.dir: hdfs://fgedu-nn:8020/bigdata/fgdata/flink/checkpoints
state.savepoints.dir: hdfs://fgedu-nn:8020/bigdata/fgdata/flink/savepoints

execution.checkpointing.interval: 60000
execution.checkpointing.timeout: 600000

high-availability: zookeeper
high-availability.storageDir: hdfs://fgedu-nn:8020/bigdata/fgdata/flink/ha
high-availability.zookeeper.quorum: fgedu-zk01:2181,fgedu-zk02:2181,fgedu-zk03:2181

rest.port: 8081
env.log.dir: /bigdata/fgdata/logs/flink
EOF

# 3. 配置masters
cat > /bigdata/app/flink/conf/masters << ‘EOF’
fgedu-jm01:8081
fgedu-jm02:8081
EOF

# 4. 配置workers
cat > /bigdata/app/flink/conf/workers << ‘EOF’
fgedu-tm01
fgedu-tm02
fgedu-tm03
fgedu-tm04
EOF

# 5. 创建目录
mkdir -p /bigdata/fgdata/flink
mkdir -p /bigdata/fgdata/logs/flink
hdfs dfs -mkdir -p /bigdata/fgdata/flink/checkpoints
hdfs dfs -mkdir -p /bigdata/fgdata/flink/savepoints
hdfs dfs -mkdir -p /bigdata/fgdata/flink/ha

# 6. 启动Flink集群
/bigdata/app/flink/bin/start-cluster.sh

# 7. 验证集群
# 访问Web UI: http://fgedu-jm01:8081
# 查看TaskManager
/bigdata/app/flink/bin/flink list

// pom.xml
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>1.17.1</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java</artifactId>
<version>1.17.1</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka</artifactId>
<version>1.17.1</version>
</dependency>

// FgeduFlinkJob.java
package com.fgedu.flink;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

public class FgeduFlinkJob {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

env.enableCheckpointing(60000);

KafkaSource<String> source = KafkaSource.<String>builder()
.setBootstrapServers(“fgedu-kafka01:9092,fgedu-kafka02:9092”)
.setTopics(“fgedu_user_events”)
.setGroupId(“fgedu_flink_group”)
.setStartingOffsets(OffsetsInitializer.latest())
.setValueOnlyDeserializer(new SimpleStringSchema())
.build();

DataStream<String> stream = env.fromSource(
source,
WatermarkStrategy.noWatermarks(),
“Kafka Source”
);

DataStream<String> result = stream
.map(new MapFunction<String, String>() {
@Override
public String map(String value) throws Exception {
return value.toUpperCase();
}
})
.keyBy(value -> value)
.window(TumblingProcessingTimeWindows.of(Time.seconds(60)))
.reduce((v1, v2) -> v1 + “,” + v2);

result.print();

env.execute(“Fgedu Flink Job”);
}
}

# 编译打包
mvn clean package

# 提交作业
/bigdata/app/flink/bin/flink run \
-c com.fgedu.flink.FgeduFlinkJob \
-p 8 \
fgedu-flink-job.jar

# 查看作业
/bigdata/app/flink/bin/flink list

# 取消作业
/bigdata/app/flink/bin/flink cancel <job-id>

# 从Savepoint恢复
/bigdata/app/flink/bin/flink run \
-s hdfs://fgedu-nn:8020/bigdata/fgdata/flink/savepoints/savepoint-xxx \
-c com.fgedu.flink.FgeduFlinkJob \
fgedu-flink-job.jar

// Checkpoint配置
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 启用Checkpoint
env.enableCheckpointing(60000); // 60秒一次

// Checkpoint模式：EXACTLY_ONCE
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

// Checkpoint超时时间
env.getCheckpointConfig().setCheckpointTimeout(600000); // 10分钟

// 最小间隔
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000); // 5秒

// 最大并发Checkpoint数
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

// 外部化Checkpoint
env.getCheckpointConfig().setExternalizedCheckpointCleanup(
ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
);

// State Backend配置
env.setStateBackend(new RocksDBStateBackend(
“hdfs://fgedu-nn:8020/bigdata/fgdata/flink/checkpoints”,
true // 增量Checkpoint
));

// RocksDB配置
RocksDBStateBackend backend = new RocksDBStateBackend(…);
backend.setDbStoragePath(“/bigdata/fgdata/flink/rocksdb”);
backend.setNumberOfTransferThreads(4);
env.setStateBackend(backend);

// 创建Savepoint
// 方式1：命令行
/bigdata/app/flink/bin/flink savepoint <job-id> \
hdfs://fgedu-nn:8020/bigdata/fgdata/flink/savepoints

// 方式2：REST API
curl -X POST http://fgedu-jm:8081/jobs/<job-id>/savepoints \
-d ‘{“target-directory”: “hdfs://fgedu-nn:8020/bigdata/fgdata/flink/savepoints”}’ \
-H “Content-Type: application/json”