0


Flink架构底层原理详解:案例解析(43天)

系列文章目录

一、Flink架构(掌握)
二、Flink代码案例(掌握)
三、UDF(熟悉)
四、Flink常见面试题整理

文章目录


前言

本文主要详解了Flink架构,通过案例详解Flink流式开发。


提示:以下是本篇文章正文内容,下面案例可供参考

一、Flink架构(掌握)

1、系统架构

官网的架构图如下:

在这里插入图片描述

讲义的架构如下:

在这里插入图片描述

1.1 通信(了解)

Spark的通信:在1.6版本及之前,用的是akka通信框架,在1.6之后,用的是netty。

在这里插入图片描述

Flink的通信:akka通信框架。

在这里插入图片描述

1.2 JobManager

作用:管理众多的TaskManager从节点。负责任务分配和资源管理

JobManager中包括如下3个组件:

  • ResourceManager:这是Flink自己的资源管理器。要和Yarn的ResourceManager区分开来。
  • JobMaster:作业调度器。负责向资源管理器申请资源;分配任务给到TaskManager进行执行
  • Dispatcher:分发器。用来接收Client进程提交的Flink任务,然后去启动JobMaster,将Flink任务转发给JobMaster
1.3 TaskManager

作用:接收JobManager分配过来的任务;同时向JobManager汇报Task执行状态、心跳等信息

1.4 Scheduler

Spark中的调度器:DAGScheduler和TaskScheduler

  • DAGScheduler:将Job任务形成DAG有向无环图和划分Stage阶段,确定每个Stage阶段有多少个Task线程
  • TaskScheduler:将DAGScheduler发送过来的TaskSet中的Task线程任务分配给到Executor进程进行执行

Flink:JobMaster作业调度器。负责向资源管理器申请资源;分配任务给到TaskManager进行执行

1.5 Checkpoint Coordinator

检查点协调器。主要负责Checkpoint的操作,对Flink程序进行容错。

1.6 Memory & IO Manager

内存和IO管理器。负责TaskManager的内存和IO管理

1.7 Network Manager

网络管理器,负责不同节点间的Slot进行数据的交换。分为如下3种场景:

# 场景1: 同一个节点,同一个TaskManager的不同Slot间
举例: 你和你的同学都在广州黑马的218教室学习
数据交换效率最高,而且不需要经过网络管理器

# 场景2: 同一个节点,不同的TaskManager的Slot间
举例: 你和你的同学都在广州黑马,但是在不同教室
数据交换效率中等,而且不需要经过网络管理器

# 场景3: 不同节点Slot间
举例: 你在广州黑马,你的同学在深圳黑马
数据交换效率最低,而且需要经过网络管理器
1.8 Client

只是负责任务的提交。提交成功后,其实可以断开了。在命令提交任务时,可以指定

-d

参数来配置。

如果配置了

-d

,则说明客户端和集群断开了。

2、任务提交流程

2.1 抽象提交流程

在这里插入图片描述

在这里插入图片描述

1- Flink任务(App)通过Client客户端提交给到分发器
2- 分发器接收到Flink任务以后,接着去启动JobManager中的JobMaster,并且将Flink任务提交给到JobMaster
3- JobMaster接收到Flink任务以后,向ResourceManager资源管理器申请Slot资源
4- 资源管理器接收到资源申请之后,首先启动新的TaskManager
5- 新的TaskManager启动以后,会反向注册回资源管理器,并且告诉它我目前有多少Slot的资源
6- 资源管理器命令TaskManager将空闲的Slot资源提供出来
7- TaskManager接收到资源提供的命令以后,将资源给到JobMaster
8- JobMaster申请到资源以后,将任务分配给到具体的TaskManager进行执行
2.2 Standalone模式提交流程

在这里插入图片描述

(1)客户端提交任务到Dispacher(分发器)

(2)Dispacher分发器启动JobMaster

(3)JobMaster启动后,它会向JobManager的ResourceManager(资源管理器)请求资源(slot)

(4)JobManager的ResourceManager(资源管理器)向TaskManager请求资源(slot)

(5)TaskManager会向JobMaster提供资源(slot)

(6)JobMaster收到资源后,会向TaskManager提交(分发)任务

(7)TaskManager收到任务后,就在Slot上执行

(8)任务执行完后,释放资源

注意:Standalone模式下,Slot资源使用完了以后,那么无法继续提交Flink程序,会报错。

/export/server/flink/bin/flink run -py /export/software/checkpoint_demo.py

在这里插入图片描述

2.3 Yarn-session模式提交流程

如果需要把任务提交在Yarn-Session下运行,则分为2步:

  • 初始化Yarn-session集群
  • 提交任务

首先看第一步。

2.3.1 初始化Session集群

(1)请求Yarn的ResourceManager(资源管理器)

(2)Yarn的ResourceManager收到请求后,会启动一个Container(容器),当然这个容器就是ApplicationMaster(AppMaster)

(3)这个AppMaster就是Flink的JobManager,这个JobManager会初始化Dispacher和ResourceManager(资源管理器)

这里还没有初始化TaskManager,因此集群没有slot资源

在这里插入图片描述

2.3.2 提交任务

在这里插入图片描述

(1)客户端提交任务给JobManager(AppMaster)的分发器(Dispacher)

(2)分发器收到任务后,会启动JobMaster

(3)JobMaster启动后,会向JobManager(AppMaster)请求资源(slot)

(4)JobManager会向Yarn的ResourceManager请求资源

(5)Yarn的ResourceManager收到请求后,会在闲置的节点动态启动Container(TaskManager)

(6)Container启动成功后,会注册给AppMaster(JobManager)的ResourceManager

(7)Container会向AppMaster(JobManager)的JobMaster提供资源(slot)

(8)JobMaster会把任务分发给Container(TaskManager)去执行

(9)待任务执行完后,Container(TaskManager)会被AppMaster(JobManager),最终留下JobManager,这个不会被销毁

2.4 Yarn-per-job模式提交流程

在这里插入图片描述

(1)客户端提交任务给Yarn的ResourceManager

(2)Yarn的ResourceManager收到请求后,会启动一个Container(AppMaster),这个AppMaster就是Flink的JobManager

(3)JobManager里有任务调度器和资源管理器,任务调度器就会开始调度任务,向JobManager的资源管理器申请资源

(4)JobManager的资源管理器它会向Yarn的ResourceManager申请资源

(5)Yarn的ResourceManager会动态启动Container(TaskManager),这些Container就是资源

(6)这些Container启动后,会反向注册给AppMaster(JobManager)

(7)这些Container向JobMaster提供资源

(8)JobMaster收到资源后,把任务分发给Container(TaskManager)去执行

(9)任务执行完后,AppMaster(JobManager)会把Container(TaskManager)注销

(10)AppMaster(JobManager)会向Yarn的ResourceManager注销自己

2.5 Yarn-application模式提交流程

与Yarn-per-job的区别是Client进程运行的地方不一样。application模式是在集群中随机找一个从节点启动和运行Client进程。Flink程序的提交流程与Yarn-per-job完全一样。

3、一些重要的概念

3.1 程序流程图

在这里插入图片描述
在这里插入图片描述

3.2 一些概念
  • 层级关系

Spark层级关系:Spark的应用 > Job任务 > DAG有向无环图 > Stage阶段 > Task线程任务

Flink层级关系:Flink的应用 > Job任务 > DAG有向无环图 > 算子链 > Task线程任务 > SubTasks子任务

  • 并行度

运行同时运行的任务数。Flink的并行度的设置如下:

#1.默认,在配置文件中,优先级最低。不推荐使用
在flink-conf.yaml中可配置

#2.任务提交时指定(推荐)
bin/flink run -p3 xxxx.jar

#3.在全局代码中配置
env.setParallelism(1)#4.在算子中,优先级最高...reduce().setParllelism(1)
  • 算子&算子链

算子:每一个对数据处理的方法/API都称之为算子。

算子链:把窄依赖的算子合并在一起。算子链能够提升数据处理效率

在这里插入图片描述

  • 宽依赖&窄依赖

Spark

宽依赖:Shuffle Dependency

窄依赖:Narrow Dependency

Flink

宽依赖(重分区):redistributing dependency

窄依赖(一对一):one-to-one dependency

  • 概念

Job:Flink的程序

Task:Flink的并行度

SubTask:每个任务中的子任务数

  • Slot槽&槽共享

在这里插入图片描述

槽:slot,是集群的静态资源,在Standalone模式下,槽是预先配置的,不能更改。如果要改,改完后需要重启集群。

Yarn模式,可以通过启动多个TaskManager来动态初始化多个slot槽。

slot是运行Flink的单位。Flink任务必须运行在slot里。

slot和并行度是有关联的。并行度的数量不能超过可用slot的数量。

槽共享:一个槽可以运行不同Task下的多个SubTask。

不同的Task下的相同SubTask,尽量在同一个slot上执行,这是为了提升程序的执行效率。这就是槽共享
相同的Task下的SubTask,一定不会在同一个slot上执行,这是为了充分利用集群资源,达到并行效果。

二、Flink代码案例(掌握)

1、需求

使用代码来实现Flink的wordcount案例。
SparkCore版的WordCount实现过程:
1- 创建顶级对象SparkContext
2- 数据输入
3- 数据处理
    3.1- 文本内容切分: flatMap
    3.2- 数据格式转换: map
    3.3- 分组聚合: reduceByKey
4- 数据输出
5- 释放资源

2、Flink流式程序开发流程

1- 创建流式执行环境
2- 数据输入
3- 数据处理
4- 数据输出
5- 启动流式任务
Flink中算子的分类:
1- source算子: 数据读取
2- transformation算子: 数据处理
3- sink算子: 数据输出

3、创建项目

前提条件:无论是在远程Linux环境还是本地Windows环境。要想成功开发Python版Flink,都需要有Python环境。

推荐如下的操作,在虚拟机集群的所有节点上都执行一次:

#1.保证有Python3.6、3.7或者3.8
python -V#2.安装flink依赖
python -m pip install apache-flink==1.15.4 -i https://pypi.tuna.tsinghua.edu.cn/simple

在这里插入图片描述

4、实现

https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/python/datastream_tutorial/

https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/python/table_api_tutorial/

在这里插入图片描述

4.1 批案例
from pyflink.datastream import StreamExecutionEnvironment, RuntimeExecutionMode
import os

os.environ['JAVA_HOME']='/export/server/jdk1.8.0_241'
os.environ['PYSPARK_PYTHON']='/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON']='/root/anaconda3/bin/python3'if __name__ =='__main__':# 1 - 创建流式执行环境# 1.1- 得到顶级对象
    env = StreamExecutionEnvironment.get_execution_environment()# 1.2- 设置运行模式:批处理
    env.set_runtime_mode(RuntimeExecutionMode.BATCH)# 1.3- 设置并行度:全局并行度
    env.set_parallelism(1)# 2 - 数据输入
    init_ds = env.read_text_file(file_path="file:///export/data/flink_base/content.txt",charset_name="UTF-8")# 3 - 数据处理# 3.1- 将文本内容切分得到一个个单词"""
        lambda 形参1,形参2... : 单行代码
    """
    flatmap_ds = init_ds.flat_map(lambda line: line.split(" "))# 3.2- 将单词转成元组
    map_ds = flatmap_ds.map(lambda word:(word,1))# 3.3- 按照单词分组
    keyby_ds = map_ds.key_by(lambda tup: tup[0])# 3.4- 对单词的次数进行聚合"""
        rdd.reduceByKey(lambda agg,curr: agg+curr)
    """# 错误代码# result = keyby_ds.reduce(lambda agg,curr: agg+curr)

    result = keyby_ds.reduce(lambda tup1,tup2:(tup1[0],tup1[1]+tup2[1]))# 4 - 数据输出
    result.print()# 5 - 启动流式任务
    env.execute()

运行结果截图:

在这里插入图片描述

可能遇到的错误:

在这里插入图片描述

原因: 1- 服务器上没有安装JDK;2- 安装了JDK,但是在代码中没有明确告诉程序JDK在什么地方
解决办法: 在flink代码文件上面添加如下内容
import os
os.environ['JAVA_HOME'] = '/export/server/jdk1.8.0_241'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'
4.2 流案例
from pyflink.datastream import StreamExecutionEnvironment, RuntimeExecutionMode, DataStream
import os

os.environ['JAVA_HOME']='/export/server/jdk1.8.0_241'
os.environ['PYSPARK_PYTHON']='/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON']='/root/anaconda3/bin/python3'if __name__ =='__main__':# 1 - 创建流式执行环境# 1.1- 得到顶级对象
    env = StreamExecutionEnvironment.get_execution_environment()# 1.2- 设置运行模式:流处理
    env.set_runtime_mode(RuntimeExecutionMode.STREAMING)# 1.3- 设置并行度:全局并行度
    env.set_parallelism(1)# 2 - 数据输入
    init_ds = DataStream(env._j_stream_execution_environment.socketTextStream("192.168.88.161",9999))# 3 - 数据处理# 3.1- 将文本内容切分得到一个个单词"""
        lambda 形参1,形参2... : 单行代码
    """
    flatmap_ds = init_ds.flat_map(lambda line: line.split(" "))# 3.2- 将单词转成元组
    map_ds = flatmap_ds.map(lambda word:(word,1))# 3.3- 按照单词分组
    keyby_ds = map_ds.key_by(lambda tup: tup[0])# 3.4- 对单词的次数进行聚合"""
        rdd.reduceByKey(lambda agg,curr: agg+curr)
    """# 错误代码# result = keyby_ds.reduce(lambda agg,curr: agg+curr)

    result = keyby_ds.reduce(lambda tup1,tup2:(tup1[0],tup1[1]+tup2[1]))# 4 - 数据输出
    result.print()# 5 - 启动流式任务
    env.execute()

运行结果截图:

在这里插入图片描述

可能遇到的错误:

在这里插入图片描述

原因: 9999端口号没有启动
解决办法: 提前在程序运行前,在node1上执行nc -lk 9999
4.3 SQL案例
import os

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment

os.environ['FLINK_HOME']='/export/server/flink'
os.environ['JAVA_HOME']='/export/server/jdk1.8.0_241'
os.environ['PYSPARK_PYTHON']='/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON']='/root/anaconda3/bin/python3'if __name__ =='__main__':# 1- 创建流式执行任务# 1.1- 创建顶级对象
    env = StreamExecutionEnvironment.get_execution_environment()# 1.2- 得到FlinkSQL层级的顶级对象
    table = StreamTableEnvironment.create(stream_execution_environment=env)# 1.3- 设置并行度:全局并行度# env.set_parallelism(1)# 2- 数据输入
    table.execute_sql("""
        create table source(
            word varchar
        ) with (
            'connector'='socket',
            'hostname'='192.168.88.161',
            'port'='9999',
            'format'='csv'
        )
    """)# 3- 数据输出
    table.execute_sql("""
        create table sink(
            word varchar,
            cnt bigint
        ) with (
            'connector'='print'
        )
    """)# 4- 数据处理
    table.execute_sql("""
        insert into sink
        select 
            word,
            count(1) as cnt
        from source
        group by word
    """).wait()# 5- 启动流式任务
    env.execute()

运行结果截图:

在这里插入图片描述

可能遇到的错误一:

在这里插入图片描述

原因: 代码没有找到Flink的安装目录在什么地方
解决办法: 在代码的上面添加如下内容
os.environ['FLINK_HOME'] = '/export/server/flink'

可能遇到的错误二:

在这里插入图片描述

原因: 需要在增删改查的语句代码后面增加wait()的方法调用

在这里插入图片描述

5、提交运行

5.1 开源提交

环境准备:在node1上执行即可

需要确保Flink的Standalone集群是启动的状态,如果没有启动,需要执行如下命令:
cd /export/server/flink/bin
./start-cluster.sh

启动nc
nc -lk 9999

在这里插入图片描述

提交命令:在node1上执行即可

/export/server/flink/bin/flink run -py /export/data/flink_base/flink_sql_wordcount.py

注意: 代码所在的路径要改成你自己的

运行成功截图如下:

在这里插入图片描述

在这里插入图片描述

5.2 阿里云提交

S

在这里插入图片描述

在这里插入图片描述

在这里插入图片描述

在这里插入图片描述

在这里插入图片描述
)

运行结果截图:

在这里插入图片描述

三、UDF(熟悉)

1、概述

UDF,user defined function,用户自定义函数。

官网如下:

https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/table/functions/udfs/

https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/python/table/udfs/python_udfs/

Flink的UDF函数可以分为如下几种类型:

  • Scalar Function:标量函数,UDF,一对一。举例:split、substr、concat
  • Table Function:表数据生成函数,UDTF,一对多。举例:explode、json_tuple
  • Aggregate Function:聚合函数,UDAF,多对一。举例:sum、avg、max、min、count等
  • Table Aggregate Function:表数据生成聚合函数,UDTAF,多对多。

2、Scalar Function

Scalar Function,UDF。就是一进一出的函数。比如map方法。

2.1 需求
实现一个类似于两数之和的sum函数,函数名:mySum
优先采用SQL来实现。

输入数据:
| num1 | num2 |
|  1   |   2  |
|  3   |   4  |

输出结果:
| num1 | num2 | result |
|  1   |   2  |   3    |
|  3   |   4  |   7    |
2.2 实现
import os

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, DataTypes
from pyflink.table.udf import udf

os.environ['FLINK_HOME']='/export/server/flink'
os.environ['JAVA_HOME']='/export/server/jdk1.8.0_241'
os.environ['PYSPARK_PYTHON']='/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON']='/root/anaconda3/bin/python3'"""
    实现一个类似于两数之和的sum函数,函数名:mySum
"""if __name__ =='__main__':# 1- 创建流式执行环境# 1.1- 创建顶级对象
    env = StreamExecutionEnvironment.get_execution_environment()# 1.2- 得到SQL API层的顶级对象
    table = StreamTableEnvironment.create(stream_execution_environment=env)# 1.3- 设置全局并行度
    env.set_parallelism(1)# 2- 数据输入
    table.execute_sql("""
        create table source(
            num1 bigint,
            num2 bigint
        ) with (
            'connector'='socket',
            'hostname'='192.168.88.161',
            'port'='9999',
            'format'='csv'
        )
    """)# 3- 数据输出
    table.execute_sql("""
        create table sink(
            num1 bigint,
            num2 bigint,
            `result` bigint
        ) with (
            'connector'='print'
        )
    """)# 4- 数据处理# 4.1- 创建自定义Python函数@udf(result_type=DataTypes.BIGINT())defmySum_func(num_arg1, num_arg2):return num_arg1 + num_arg2

    # 4.2- 注册
    table.create_temporary_function('mySum',mySum_func)# 4.3- 调用
    table.execute_sql("""
        insert into sink
        select
            num1,num2,mySum(num1,num2) as  `result`
        from source
    """).wait()# 5- 启动流式任务
    env.execute()

运行结果截图:

在这里插入图片描述

3、Table Function

Table Function,表值函数,一进多出的函数。类似于Hive中的UDTF。

3.1 需求
实现一个类似于flatMap的功能(explode)的功能。数据源来自于socket。函数名:myExplode。

输入数据:
| mynum |
|  3    |
|  4    |

输出结果: 返回<mynum,并且大于等于0的数字
| result |
|   0    |
|   1    |
|   2    |

| result |
|   0    |
|   1    |
|   2    |
|   3    |
3.2 实现
udtf在SQL语句中进行调用,语法比较特殊
格式: lateral table(UDTF函数名称(字段名称)) as 视图名称(视图中新的字段名称1,视图中新的字段名称2..视图中新的字段名称n) on true
import os

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, DataTypes
from pyflink.table.udf import udf,udtf

os.environ['FLINK_HOME']='/export/server/flink'
os.environ['JAVA_HOME']='/export/server/jdk1.8.0_241'
os.environ['PYSPARK_PYTHON']='/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON']='/root/anaconda3/bin/python3'"""
    实现一个类似于flatMap的功能(explode)的功能。数据源来自于socket。函数名:myExplode
    UDTF
"""if __name__ =='__main__':# 1- 创建流式执行环境# 1.1- 创建顶级对象
    env = StreamExecutionEnvironment.get_execution_environment()# 1.2- 得到SQL API层的顶级对象
    table = StreamTableEnvironment.create(stream_execution_environment=env)# 1.3- 设置全局并行度
    env.set_parallelism(1)# 2- 数据输入
    table.execute_sql("""
        create table source(
            mynum bigint
        ) with (
            'connector'='socket',
            'hostname'='192.168.88.161',
            'port'='9999',
            'format'='csv'
        )
    """)# 3- 数据输出
    table.execute_sql("""
        create table sink(
            `result` bigint
        ) with (
            'connector'='print'
        )
    """)# 4- 数据处理# 4.1- 创建自定义的Python函数@udtf(result_types=DataTypes.BIGINT())defmyExplode_func(num_arg):returnrange(num_arg)# 4.2- 注册进Flink中# 下面2种任意使用其中一个都行
    table.create_temporary_system_function('myExplode', myExplode_func)# table.create_temporary_function('myExplode', myExplode_func)# 4.3- 调用# 错误调用# table.execute_sql("""#     insert into sink#     select#         myExplode(mynum) as `result`#     from source# """).wait()# 正确调用"""
        udtf在SQL语句中进行调用,语法比较特殊
        格式:lateral table(UDTF函数名称(字段名称)) as 视图名称(视图中新的字段名称1,视图中新的字段名称2..视图中新的字段名称n) on true
    """
    table.execute_sql("""
        insert into sink
        select
            new_field
        from source
        left join lateral table(myExplode(mynum)) as tt(new_field) on true
        -- Hive中UDTF函数调用:lateral view explode(split(line,' ')) t as new_field
    """).wait()# 5- 启动流式任务
    env.execute()

运行结果截图:

在这里插入图片描述

可能遇到的错误一:

在这里插入图片描述

原因: result单词在Flink的SQL中是一个关键字
解决办法:
    1- 修改result字段的名称,变成不是关键字的
    2- 在result上面加上``

可能遇到的错误二:

在这里插入图片描述

原因: UDTF的调用在FlinkSQL中有特殊的语法要求
解决办法: 改成如下的SQL语句
insert into sink
select
    new_field
from source
left join lateral table(myExplode(mynum)) as tt(new_field) on true

可能遇到的错误三:

在这里插入图片描述

原因: 对UDTF进行注册需要使用@udtf装饰器

可能遇到的错误四:

在这里插入图片描述

原因: @udtf装饰器中的参数名叫做  result_types

4、Aggregate Function

Aggregate Function,聚合函数,是多进一出的函数,类似于Hive的UDAF函数。

4.1 需求
实现一个类似于count的函数,统计词频。数据源为socket,函数名:myCount

输入数据:
|  word  |
| hello  |
| spark  |
| hello  |

输出数据:
| word  |   cnt  |
| hello |    2   |
| spark |    1   |
4.2 实现
import os

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, DataTypes
from pyflink.table.types import DataType
from pyflink.table.udf import udf, AggregateFunction, ACC, T

os.environ['FLINK_HOME']='/export/server/flink'
os.environ['JAVA_HOME']='/export/server/jdk1.8.0_241'
os.environ['PYSPARK_PYTHON']='/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON']='/root/anaconda3/bin/python3'"""
    实现一个类似于count的函数,统计词频。数据源为socket,函数名:myCount
"""# 4.1- 创建udaf函数的相关代码classMyUDAF(AggregateFunction):"""
        创建累加器,也就是用来对数据进行初始化
        累加器的作用:用来临时保存中间的聚合结果
    """defcreate_accumulator(self)-> ACC:# 返回列表,里面的0表示的是单词次数的初始值return[0]# 从累加器中获取数据defget_value(self, accumulator: ACC)-> T:# 这里的0是索引/下标return accumulator[0]# 对数据进行累加。+Udefaccumulate(self, accumulator: ACC,*args):# 相同的单词,每来一个就对单词次数+1# accumulator[0] = accumulator[0] + 1
        accumulator[0]+=1# 撤回累加器数据的变化。-Udefretract(self, accumulator: ACC,*args):
        accumulator[0]= accumulator[0]-1# 合并多个累加器中的值,因为Flink是多线程分布式运行defmerge(self, accumulator: ACC, accumulators):for i in accumulators:
            accumulator[0]= accumulator[0]+ i[0]# 获得最终结果的数据类型defget_result_type(self)-> DataType:return DataTypes.BIGINT()# 获得累加器中存放的元素的数据类型defget_accumulator_type(self)-> DataType:return DataTypes.BIGINT()if __name__ =='__main__':# 1- 创建流式执行环境# 1.1- 创建顶级对象
    env = StreamExecutionEnvironment.get_execution_environment()# 1.2- 得到SQL API层的顶级对象
    table = StreamTableEnvironment.create(stream_execution_environment=env)# 1.3- 设置全局并行度
    env.set_parallelism(1)# 2- 数据输入
    table.execute_sql("""
        create table source(
            word varchar
        ) with (
            'connector'='socket',
            'hostname'='192.168.88.161',
            'port'='9999',
            'format'='csv'
        )
    """)# 3- 数据输出
    table.execute_sql("""
        create table sink(
            word varchar,
            cnt bigint
        ) with (
            'connector'='print'
        )
    """)# 4- 数据处理# 4.2- 注册进Flink"""
        UDAF注册的时候,传递的是类的实例对象,也就是类名(参数)
    """
    table.create_temporary_function('myCount',MyUDAF())# 4.3- 调用
    table.execute_sql("""
        insert into sink
        select
            word,
            myCount(1) as cnt
        from source
        group by word
    """).wait()# 5- 启动流式任务
    env.execute()

运行结果截图:

在这里插入图片描述

5、阿里云UDF

在进行函数注册时,先把函数开发好。

开发阿里云的UDF参考手册:https://help.aliyun.com/zh/flink/developer-reference/python/?spm=a2c4g.11186623.0.0.2ac522158vB92w

5.1 注册UDF函数

选择SQL开发 -> 函数选项,上传压缩包,如下图:

在这里插入图片描述

点击确定,如下图:

在这里插入图片描述

点击创建函数,提示创建成功,如下图:

在这里插入图片描述

到此,则函数创建成功。

5.2 使用UDF函数

在这里插入图片描述

在这里插入图片描述

  • sub_string函数
#1.创建表
CREATE TEMPORARY TABLE function_udf(
  a VARCHAR,
  b INT,
  c INT
) WITH ('connector'='socket',
  'hostname'='172.24.24.49',
  'port'='9999',
  'format'='csv');#2.查询SQL
SELECT sub_string(a,2,5) FROM function_udf;

注意: hostname一定要改成自己的ECS服务器内网IP

在这里插入图片描述

  • split函数
#1.创建表
同上,略

#2.查询SQL
SELECT a,b,c,d,e
FROM function_udf,lateral table(split(a)) as T(d,e);

在这里插入图片描述

  • weight_avg函数
#1.创建表
同上,略。

#2.查询SQL
SELECT weighted_avg(b,c) FROM function_udf;

在这里插入图片描述

四、Flink常见面试题整理

1、Flink中的部署模式?你是如何部署Flink?你的Flink的项目是用什么方式部署?

  • 开源Flink:我们使用的是开源版的Flink,部署项目的时候使用的Application应用模式,给你具体说下为什么我们使用application模式进行部署,xxxx。另外我给您介绍下其他一些部署模式。Session、per-job
  • 阿里云Flink:我们使用的是阿里云Flink,部署项目的时候使用的per-job,也就是job分离模式,给你具体说下为什么我们使用job分离模式进行部署,xxxx。另外我给您介绍下其他一些部署模式。Session、application

2、说一下对Flink中时间的理解?你在你的项目中是如何使用Flink进行数据统计的?

步骤一:介绍Flink中的时间分类,并且要说出每个时间代表的含义。

步骤二:结合业务举例说明如何使用3类时间。

​ 举例:在项目中,我们使用Flink进行数据的实时ETL。但是在做数据检查和核对的时候,发现有些数据出现事件时间的缺失,排错问题发现是业务方上报数据,导致了部分事件时间的空缺。然后与业务方进行异常数据处理方式的沟通,最终确定是使用处理时间来填补事件时间空缺的情况。

3、你在项目中如何解决延迟到来的数据?如何彻底解决数据延迟到的情况?

在Flink的SQL开发中,无法解决数据超过watermark允许延迟时间后到来的数据被丢失的情况。在DataStream的开发中可以通过侧输出流解决延迟来的数据,也就是在DataStream的编程中,可以做到彻底的解决数据延迟到来的情况。

如何彻底解决数据延迟到的情况?watermark水印+侧输出流

在这里插入图片描述

4、介绍一下Flink的底层原理?介绍一下Flink的架构?

步骤一:从Flink的作业提交运行时的架构说起:我们开发完Flink程序以后,通过命令行或者界面提交Flink程序到集群中运行,首先第一步会启动Client客户端进程。接着Client进程将我们的Flink job通过Actor通信系统提交给JobManager,JobManager拿到任务后会分配给到具体的TaskManager来执行,并且任务运行的具体场所是TaskManager中的Slot

步骤二:我们的项目中,使用的是阿里云Flink。因此我们通过per-job模式部署Flink程序。接下来给您具体介绍下该模式下任务的底层提交流式是什么样的?

步骤三:回答Per-job作业的提交流程

5、用户自定义函数的分类

UDF:用户自定义函数,输入一行数据,得到一行数据。一对一

UDAF:用户自定义聚合函数,输入多行数据,得到一行数据。多对一

UDTF:用户自定义表数据生成函数,输入一行数据,得到多行数据。一对多

标签: flink 架构 大数据

本文转载自: https://blog.csdn.net/syhiiu/article/details/140587838
版权归原作者 大数据小朋友 所有, 如有侵权,请联系我们删除。

“Flink架构底层原理详解:案例解析(43天)”的评论:

还没有评论