Apache Flink是一个在有界数据流和无界数据流上进行有状态计算分布式处理引擎和框架。Flink 设计旨在所有常见的集群环境中运行，以任意规模和内存级速度执行计算。
一、主要特点和功能
1. 实时流处理:
- 低延迟: Flink 能够以亚秒级的延迟处理数据流，非常适合对时间敏感的应用，如实时分析、监控和告警系统。
- 状态管理: Flink 提供了对状态的内置支持，使得开发有状态的流式处理应用变得容易，如窗口操作、复杂事件处理等。
2. 批处理和流处理的统一:
-
Flink 既可以用于流处理，也可以用于批处理，允许用户在一个框架中编写应用程序，而不必在批处理和流处理之间切换。
-
事件时间处理: Flink 支持事件时间语义，可以基于数据本身的时间戳进行处理，而不是数据到达的时间，这对于处理无序数据流非常重要。
// scala table api // 引入 Flink 的批处理环境 val env = ExecutionEnvironment.getExecutionEnvironment //批处理: 读取文本文件 env.readTextFile("data/words.txt") // 处理数据: 切换、转换、分组、聚合 .flatMap(_.replaceAll("[^a-zA-Z ]", "").split("\\s+")) .map((_, 1)) .groupBy(0) .sum(1) // 输出 .print()// 引入 Flink 的流处理环境 val env = StreamExecutionEnvironment.getExecutionEnvironment // 有界流处理: 文件数据 env.readTextFile("data/words.txt") .flatMap(_.replaceAll("[^a-zA-Z ]", "").split("\\s+")) .map((_, 1)) .keyBy(_._1) .sum(1) .print() // 无界流处理: Socket env.socketTextStream("nodeIp", 9999) .flatMap(_.replaceAll("[^a-zA-Z ]", "").split("\\s+")) .map((_, 1)) .keyBy(_._1) .sum(1) .print() // 执行 Flink 作业，并给它命名 env.execute("Word Count Example")
3. 高度可扩展性:
- Flink 能够在大规模分布式集群上运行，处理从几千到上百万个事件每秒的数据流。
- 弹性和容错: Flink 使用检查点和保存点机制来提供容错能力，确保在发生故障时可以从之前的状态恢复，减少数据丢失。
4. 支持多种数据源和接收器:
- Flink 能够与多种数据源和接收器集成，如 Kafka、HDFS、Cassandra、Elasticsearch 等，使其可以轻松地处理和存储来自不同系统的数据。
5. 丰富的 API 和库:
- DataStream API: 用于流处理，允许开发者定义复杂的数据流处理逻辑。
- DataSet API: 用于批处理，提供了丰富的操作符来处理静态数据集(将在 Flink 2.0 版本被删除如何迁移 DataSet 到 DataStream | Apache Flink)
- Table API 和 SQL: 提供了一个更高级别的 API，允许开发者使用 SQL 查询来处理数据流和数据集。
- 机器学习和图处理库: Flink 提供了机器学习库（FlinkML）和图处理库（Gelly），适用于高级分析任务。
6. 部署灵活性:
- Flink 可以部署在多种环境中，如独立集群、YARN、Kubernetes、Mesos 以及本地环境中。
- 流批一体: Flink 支持将批处理和流处理集成到同一个应用程序中，简化了部署和管理。
7. 社区与生态系统:
- Flink 由一个活跃的开源社区维护和发展，生态系统日益壮大，支持越来越多的第三方工具和集成。
典型应用场景
- 实时数据分析: Flink 可用于处理实时事件流，提供实时分析、告警和监控。
- 复杂事件处理: Flink 能够处理和识别复杂事件模式，用于金融监控、欺诈检测等。
- 日志处理: 可以实时处理和分析来自各种系统的日志数据，提取有价值的信息。
- 机器学习: Flink 的流处理能力可以用于实时更新机器学习模型，或在流数据上直接进行预测。
Apache Flink 适用于各种需要实时和批处理的应用程序，尤其是在处理大规模数据流时表现出色。·
二、Flink下载，集群安装配置
官方下载地址：Downloads | Apache Flink
1. 下载、解压、配置环境变量
wget https://dlcdn.apache.org/flink/flink-1.20.0/flink-1.20.0-bin-scala_2.12.tgz
tar -zxvf flink-1.20.0 -C /opt/software
vim /etc/profile.d/myenv.sh
# FLINK_HOME=...
# PAHT=$PATH:$FLINK_HOME/bin
2. 进入 Flink 配置目录：
cd $FLINK_HOME/conf
3. 编辑 masters 文件：
在文件中指定 JobManager 的主机名或 IP 地址。如果有多个 JobManager（用于高可用性），每个 JobManager 使用一行。
vim masters
# 格式如下
<JobManager1>:<port>
<JobManager2>:<port> # 如果有高可用性设置
# 示例
master01:8081
master02:8081
4. 编辑 workers 文件：
在文件中列出所有 TaskManager 的主机名或 IP 地址，每个 TaskManager 使用一行
vim workers
# 格式如下
<TaskManager1>
<TaskManager2>
# 示例
worker01
worker02
5. 编辑 flink-conf.yaml 文件：
flink-conf.yaml 是 Flink 的主要配置文件，用于配置各种集群参数。
vim flink-conf.yaml
# 指定 JobManager 的 RPC 服务监听的地址
jobmanager.rpc.address:
# 指定 JobManager 在所有网络接口上进行绑定
jobmanager.bind-host: 0.0.0.0
# 指定 TaskManager 进程的外部地址
taskmanager.host: master01 # 每台机器这里不同，其他相同
# 指定 TaskManager 绑定的网络接口
taskmanager.bind-host: 0.0.0.0
# 指定 Flink 集群中 REST API 服务的外部地址
rest.address: master01
# 指定 REST API 服务在所有网络接口上进行绑定。
rest.bind-address: 0.0.0.0
以下看需配置
# TaskManager 内存：
taskmanager.memory.process.size: 1024m
# TaskManager 的槽位数：
taskmanager.numberOfTaskSlots: 4
# 高可用性配置（如果需要高可用性）：
high-availability.type: zookeeper
high-availability.zookeeper.quorum: localhost:2181
high-availability.zookeeper.path.root: /flink
high-availability.cluster-id: /cluster_one
high-availability.storageDir: hdfs:///flink/recovery
state.checkpoints.dir: hdfs:///flink/checkpoints
state.savepoints.dir: hdfs:///flink/savepoints
# 并行度：
parallelism.default: 4
# 日志配置：
taskmanager.log.dir: /var/log/flink
# HDFS 配置：
fs.default-scheme: hdfs://namenode:9000
6. 启动 Flink 集群
$FLINK_HOME/bin/stop-cluster.sh
$FLINK_HOME/bin/start-cluster.sh
7. 提交Flink任务
