Flink 基础篇《一》

Posted on by hackdl

Apache flink是什么

框架和分布式

无界和有界数据流

有状态

flink 框架处理流程

Flink 应用场景

Apache Flink 功能强大，支持开发和运行多种不同种类的应用程序。它的主要特性包括：批流一体化、精密的状态管理、事件时间支持以及精确一次的状态一致性保障等。Flink 不仅可以运行在包括 YARN、 Mesos、Kubernetes 在内的多种资源管理框架上，还支持在裸机集群上独立部署。在启用高可用选项的情况下，它不存在单点失效问题。事实证明，Flink 已经可以扩展到数千核心，其状态可以达到 TB 级别，且仍能保持高吞吐、低延迟的特性。世界各地有很多要求严苛的流处理应用都运行在 Flink 之上。

为什么选择flink

  • 批处理和流处理
  • 流数据更真实存在
  • 目的

低延迟

高吞吐

结果准确性和良好的容错性

主要吞吐瓶颈：数据库读写(如联表操作)、数据计算。

传统数据处理架构

  • 联机事务处理OLTP（on-line transaction processing）

OLTP面向应用系统数据库（面向应用）

简单事务

实时

数据量

  • 联机分析处理OLAP（On-Line Analytical Processing）

分区技术、并行技术

OLAP是面向数据仓库（面向主题）

复杂查询

数据量大

离线

有状态的流式处理

多个架构并行时，存在数据乱序问题

流处理演变

  • lambda架构

第二代流处理架构

新一代流处理—Flink

一套系统实现lambda架构的功能。

  • 核心特点
  • 高吞吐、低延迟
  • 结果的准确性 （时间语义–>解决数据乱序）
  • 精确一次（exactly-once）的状态一致性保证
  • 常用存储系统连接
  • 高可用，动态扩展

流处理的应用场景

  • 事件驱动型应用

  • 事件分析型应用

  • 数据管道型应用

如ETL

Flink的分层API

  • 分层API

Flink vs Spark

  • 数据处理架构

spark基于批设计，流是“微批次”。

flink基于流设计，批是“有界流”。

  • 数据模型
  • spark 采用RDD模型（见备注），spark streaming的DStream本质是一组组小批RDD集合。
  • flink 基本数据模型是数据流(datastream)、事件序列(event)
  • 运行时架构
  • spark 批计算，DAG划分为不同stage，一个完成计算下一个。
  • flink 流计算，一个事件处理完成后发往下一个节点。
  • 备注

RDD可伸缩的分布式数据集（Resilient Distributed Dataset）。

Flink 快速上手

环境准备

win10

java8

idea

maven

git

flink 1.13.0

流处理

有界流(code)

WordCount任务

import org.apache.flink.stream.api.datastream.DataStreamSource

public class BoundedStreamWordCount{
    public static main(String[] args) throws Exception{
        // 1. 创建流式的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 2. 读取文件
        DataStreamSource lineDataStreamSource = env.readTextFile("input/words.txt");
        
        // 3. 转换计算
        lineDataStreamSource.flatMap((String line,Collection> out)-> {
            String[] words = line.split(" ");
            for (String word : words){
                out.collection(Tuple2.of(word,1L));
            }
        })
                .returns(Types.TUPLE(Types.STRING,Types.LONG)) // 因为JAVA类型擦除，我们给flink指明类型。
        
        // 4. 分组
        keyedStream,String> wordAndOneKeyedStream = wordAndOneTuple.keyBy(data -> data.f0); // 输入key序列
        
        // 5. 求和
        SingleOutputStreamOperator> sum = wordAndOneKeyedStream.sum(1);
        
        // 6. 打印
        sum.print();
        
        // 7. 执行  
        env.execute();

    }

}

无界流(code)

  • 有界流和无界流的差异代码
  • 有界流读取文件

DataStreamSource lineDataStreamSource = env.readTextFile("input/words.txt");

  • 无界流读取文本流(监听端口)

DataStreamSource lineDataStreamSource = env.socketTextStream("hadoop102",7777);

import org.apache.flink.stream.api.datastream.DataStreamSource

public class BoundedStreamWordCount{
    public static main(String[] args) throws Exception{
        // 1. 创建流式的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 2. 读取文本流（监听端口）
        DataStreamSource lineDataStreamSource = env.socketTextStream("hadoop102",7777);
        
        // 3. 转换计算
        lineDataStreamSource.flatMap((String line,Collection> out)-> {
            String[] words = line.split(" ");
            for (String word : words){
                out.collection(Tuple2.of(word,1L));
            }
        })
                .returns(Types.TUPLE(Types.STRING,Types.LONG)) // 因为JAVA类型擦除，我们给flink指明类型。
        
        // 4. 分组
        keyedStream,String> wordAndOneKeyedStream = wordAndOneTuple.keyBy(data -> data.f0); // 输入key序列
        
        // 5. 求和
        SingleOutputStreamOperator> sum = wordAndOneKeyedStream.sum(1);
        
        // 6. 打印
        sum.print();
        
        // 7. 执行  
        env.execute();

    }

}
// 2. 读取文本流（监听端口）
ParameterTool parameterTool = ParameterTool.fromArgs(args);
String hostname = parameterTool.get("host");
Integer port = parameteerTool.get("port");
DataStreamSource lineDataStreamSource = env.socketTextStream(hostname,port);

  • 补充：使用nc监听端口，来做作为俩天工具

1、

[atguigu@hadoop102 ~]$ nc -lk 7700

2、

[atguigu@hadoop103 ~]$ nc hadoop102 7777
你好
我是hadoop103

服务端收到数据

Flink部署

快速启动

  • Flink 主要组件

Client 提交任务

JobManager 对作业中央管理

TaskManager 具体干活的，也叫WorkerNodes

  • 环境配置

  • 本地启动

无需任何修改，bin/start-cluster.sh 即可启动

  • 集群启动

分发flink环境

配置conf/masters

配置conf/workers

bin/start-cluster.sh 启动

  • 向集群提交作业

sokcer文件流作为输出为例，nc -lk 7777

本地打包jar包

登录http://hadoop102:8081/#/submit，提交本地Jar包，设置运行参数。启动。

部署模式

  • Session Mode

先有集群（所有资源都确定了，资源共享），再提交作业

  • Per-job Mode（常用）

每一个作业启动，就启动一个集群。

  • Application Mode

每一个jar包(应用)，就启动一个集群。

独立模式 standalone

  • 会话模式部署（常用）
  • 单作业(per-job)模式部署
  • 应用模式部署

YARN 模式

flink不同版本对yarn支持不同，直接看最新的flink版本（最新版本只需要配置yarn即可）

  • 会话模式部署
  • 单作业(per-job)模式部署
  • 应用模式部署

K8S部署

…

作业提交流程

Standalone模式作业提交流程

Yarn会话模式作业提交流程

提交任务给分发器

Yarn单作业模式任务

提交任务给yarn

Flink运行时架构

DataFlow

flink的程序被映射为dataflows,包含三部分：sources、transformations、sinks。

dataflow类似有向无环图DAG，每个dataflow以任意个sources开始或任意个sinks结束。

每个算子对应一个transformation.s

Parallelism

任务并行，数据并行

conf/flink-conf.yaml文件中parallelism.default: 1

env.setParallelism(1);

sum(1).setParallelism(2)；

数据传输形式

  • 算子具有不同的并行度；算子之间的传输数据的形式有两种

One-to-one(forwarding)：stream的分区不会变，如map、fliter、flatMap算子。

Redistributing：stream的分区会发生改变，如keyBy算子基于hashCode重分区、broadcast和rebalance算子会随机分区。

  • 算子链(Operator Chains)

任务链优化技术：在特定条件下减少本地通信开销。连续的算子之间，当并行度相同且One-to-one传输模式时，fink认为它们整体一个task（原本的算子叫substask，这些substask通过local forward方式进行连接)。

执行图（ExecutionGraph）

  • StreamGraph
  • JobStream
  • ExecutionGraph
  • 物理执行图

服务器托管，北京服务器托管，服务器租用 http://www.fwqtg.net
机房租用，北京机房租用，IDC机房托管， http://www.e1idc.net

服务器托管，北京服务器托管，服务器租用，机房机柜带宽租用

服务器托管

咨询：董先生

电话13051898268 QQ/微信93663045！

上一篇:
下一篇: