Flink的详细使用

Flink的详细使用

文章目录

• Flink的详细使用
• 1. 安装部署
• • 安装
• 2. 执行任务
• • Standalone 模式
  • • 启动/停止
    • 执行任务
  • Yarn 模式
  • • Session-cluster 模式
    • • 启动 yarn-session
      • 执行任务
    • Per-Job-Cluster 模式
• 3. 执行环境
• • Environment
  • • getExecutionEnvironment（常用）
    • createLocalEnvironment
    • createRemoteEnvironment
  • Source、Sink
  • Transform（算子）
  • • map
    • flatMap
    • Filter
    • KeyBy
    • 滚动聚合算子
    • Reduce
    • Split 和 Select
    • Connect 和 CoMap
    • Union
    • 广播（broadcast）
• 4. Window
• • TimeWindow
  • CountWindow
• 5. 时间语义与 Watermark
• • 时间语义
  • Watermark（和 Window 一起使用）
  • • 乱序时间的引用
    • 顺序时间的引用
• 6. 状态管理
• • 键控状态（keyed state）
  • • Keyed State 支持数据类型
    • 例子：ValueState
• 7. ProcessFunction API
• • KeyedProcessFunction
  • TimerService 和 定时器（Timers）
  • 侧输出流（SideOutput）
• 8. 检查点（CheckPoint）
• 9. 状态一致性
• • 分类
  • 端到端 exactly-once
  • • 事务写入
    • • 预写日志（不常用）
      • 两阶段提交
  • Flink+Kafka 端到端状态一致性的保证
• Maven（pom.xml）

1. 安装部署 安装

• 第一步：将 flink-1.10.1-bin-scala_2.12.tgz 上传到服务器中并解压缩

• 第二步：修改 conf/flink-conf.yaml 文件

  # 修改 jobmanager.rpc.address 参数，修改为 jobmanager 的机器
  jobmanager.rpc.address: hadoop151
  

• 第三步：修改 conf/slaves 文件

  # slave 机器
  hadoop152
  hadoop153
  

• 第四步：将 flink 整个目录分发到其他机器上

2. 执行任务 Standalone 模式 启动/停止

• 命令

  # 启动
  bin/start-cluster.sh
  
  # 停止
  bin/stop-cluster.sh
  

• 访问 web 页面

  • http://hadoop151:8081

执行任务

# =================== 启动任务 ===================
bin/flink run -c 全限定类名 –p 分区个数 jar包
# 示例
bin/flink run -c com.itfzk.flink.wordcount.KafkaStreamWordCount -p 3 FlinkStudyDemo-1.0-SNAPSHOT-jar-with-dependencies.jar


# =================== 停止任务 ===================
bin/flink cancel JobId
# 示例
bin/flink cancel f69fbd0650ae4202b2a46b3ad2089606

Yarn 模式 Session-cluster 模式 启动 yarn-session

• 命令

  # =================== 启动 yarn-session ===================
  # -n(--container)：TaskManager 的数量
  # -s(--slots)： 每个 TaskManager 的 slot 数量，默认一个 slot 一个 core，默认每个 taskmanager 的 slot 的个数为 1，有时可以多一些 taskmanager，做冗余
  # -jm：JobManager 的内存（单位 MB)
  # -tm：每个 taskmanager 的内存（单位 MB)
  # -nm：yarn 的 appName(现在 yarn 的 ui 上的名字)
  # -d：后台执行
  bin/yarn-session.sh -n 2 -s 2 -jm 1024 -tm 1024 -nm test -d
  
  
  # =================== 停止 yarn-session ===================
  yarn application -kill Application-Id
  # 示例
  yarn application -kill application_1633171918776_0003
  

• 访问 web 页面

  • 启动 yarn-session 后会出现 web 地址，例如：http://hadoop153:42189

执行任务

# =================== 启动任务 ===================
bin/flink run -c 全限定类名 –p 分区个数 jar包
# 示例
bin/flink run -c com.itfzk.flink.wordcount.KafkaStreamWordCount -p 3 FlinkStudyDemo-1.0-SNAPSHOT-jar-with-dependencies.jar


# =================== 停止任务 ===================
bin/flink cancel JobId
# 示例
bin/flink cancel f69fbd0650ae4202b2a46b3ad2089606

Per-Job-Cluster 模式

# =================== 启动任务 ===================
bin/flink run –m yarn-cluster -c 全限定类名 –p 分区个数 jar包
# 示例
bin/flink run –m yarn-cluster -c com.itfzk.flink.wordcount.KafkaStreamWordCount -p 3 FlinkStudyDemo-1.0-SNAPSHOT-jar-with-dependencies.jar


# =================== 停止任务 ===================
bin/flink cancel JobId
# 示例
bin/flink cancel f69fbd0650ae4202b2a46b3ad2089606

3. 执行环境 Environment getExecutionEnvironment（常用）

• 创建一个执行环境，表示当前执行程序的上下文。 getExecutionEnvironment 会根据查询运行的方式决定返回什么样的运行环境，是最常用的一种创建执行环境的方式

  // 普通运行环境
  ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
  
  // 流式运行环境（常用）
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  

createLocalEnvironment

• 返回本地执行环境，需要在调用时指定默认的并行度

  LocalStreamEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(1);
  

createRemoteEnvironment

• 返回集群执行环境，将 Jar 提交到远程服务器。需要在调用时指定 JobManager的 IP 和端口号，并指定要在集群中运行的 Jar 包

  StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment("jobmanage-hostname", 6123, "YOURPATH//WordCount.jar");
  

Source、Sink

• 官网地址

Transform（算子）

• 官网地址

map

DataStream mapStram = dataStream.map(new MapFunction() {
    public Integer map(String value) throws Exception {
    }
});

flatMap

DataStream flatMapStream = dataStream.flatMap(new FlatMapFunction() {
    public void flatMap(String value, Collector out) throws Exception {
    }
});

Filter

DataStream filterStream = dataStream.filter(new FilterFunction() {
    public boolean filter(String value) throws Exception {
    }
});

KeyBy

• DataStream → KeyedStream
• 将一个流拆分成不相交的分区，每个分区包含具有相同 key 的元素，在内部以 hash 的形式实现的

滚动聚合算子

• sum()
• max()
• min()
• maxBy()
• minBy()

Reduce

• KeyedStream → DataStream
• 一个分组数据流的聚合操作，合并当前的元素和上次聚合的结果，产生一个新的值，返回的流中包含每一次聚合的结果，而不是只返回最后一次聚合的最终结果

Split 和 Select

Split

• DataStream → SplitStream
  • 根据某些特征把一个 DataStream 拆分成两个或者多个 DataStream

Select

• SplitStream→DataStream
  • 从一个 SplitStream 中获取一个或者多个DataStream

public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    String filePath = "E:\~fzk\java\IDEA\bigdata\FlinkStudyDemo\test\test1";
    DataStream inputDataStream = env.readTextFile(filePath);
    
    DataStream map = inputDataStream.map(new MapFunction() {
        public SensorsData map(String value) throws Exception {
            String[] splits = value.split(" ");
            return new SensorsData(splits[0], new Long(splits[1]), new Double(splits[2]));
        }
    });
    
    KeyedStream keyedStream = map.keyBy("id");

    // split：分流
    SplitStream splitStream = keyedStream.split(new OutputSelector() {
        public Iterable select(SensorsData value) {
            return value.getWendu() > 37 ? Collections.singletonList("h") : Collections.singletonList("d");
        }
    });

    // select：选择一个或多个 DataStream
    DataStream resultDataStream = splitStream.select("d");

    env.execute();
}

Connect 和 CoMap

Connect

• DataStream,DataStream → ConnectedStreams：连接两个保持他们类型的数据流，两个数据流被 Connect 之后，只是被放在了一个同一个流中，内部依然保持各自的数据和形式不发生任何变化，两个流相互独立

CoMap

• ConnectedStreams → DataStream：作用于 ConnectedStreams 上，功能与 map 和 flatMap 一样，对 ConnectedStreams 中的每一个 Stream 分别进行 map 和 flatMap处理

public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    String filePath = "E:\~fzk\java\IDEA\bigdata\FlinkStudyDemo\test\test1";
    DataStream inputDataStream = env.readTextFile(filePath);

    DataStream map = inputDataStream.map(new MapFunction() {
        public SensorsData map(String value) throws Exception {
            String[] splits = value.split(" ");
            return new SensorsData(splits[0], new Long(splits[1]), new Double(splits[2]));
        }
    });

    KeyedStream keyedStream = map.keyBy("id");

    SplitStream splitStream = keyedStream.split(new OutputSelector() {
        public Iterable select(SensorsData value) {
            return value.getWendu() > 37 ? Collections.singletonList("high") : Collections.singletonList("low");
        }
    });

    DataStream highDataStream = splitStream.select("high");
    DataStream lowDataStream = splitStream.select("low");

    // connect & CoMapFunction：合流
    ConnectedStreams connectedStreams = highDataStream.connect(lowDataStream);
    
    DataStream