一、概述Flink SQL 可以将多种数据源或数据落地端映射为 table 二、实现2.1. 架构设计Flink 使用 SPI 机制加载Factory(DynamicTableSourceFactory 和 DynamicTableSinkFactory同属 Factory)。在 flink-table-api-java-bridge 项目的 resources/META-INF/services 目录可以找到org.apache.flink.table.factories.Factory 文件，内容为： 123org.apache.flink.table.factories.DataGenTableSourceFactoryorg.apache.flink.table.factories.BlackHoleTableSinkFactoryorg.apache.flink.table.factories.PrintTableSinkFactory 以 DataGenTableSourceFactory 为例: 2.1.1. DynamicTableSourceFactory2. ...

一、概述123456789101112131415161718192021222324252627282930private val AVG_TEMPLATE_SINK_TABLE_SQL: String = """ |CREATE TABLE avg_template_sink_table ( | `avg_temp_val` DOUBLE, | `address` STRING, | `ts` TIMESTAMP(3) |) WITH ( | 'connector' = 'print' |) |""".stripMargin def main(args: Array[String]): Unit = { val settings: EnvironmentSettings = EnvironmentSettings.newInstance().inStreamingMode ...

一、概述为了使用 SQL，首先需要解决的是元数据管理的问题。包括 Table/UDF/View 元数据的注册、查询以及验证。 二、架构设计2.1. 元数据注册FinkSQL 原生支持了 DDL 语法，比如 CREATE 语句，通过 Catalog API 将表/视图/函数 注册到当前或指定的 Catalog 中，默认使用 InMemoryCatalog 将信息临时保存在内存中，有一个唯一的 ID，由三部分组成: 目录 (catalog)、数据库 (database) 名、表名。 在默认情况下，目录名为 defaultcatalog、数据库名为 default database。故直接创建一个 test 表，它的 ID 是: defaultcatalog.default database.MyTable 同时也提供了 HiveCatalog 与 HiveMetastore 进行集成。FlinkSQL 目前支持以下 CREATE 语句: CREATE TABLE CREATE CATALOG CREATE DATABASE CREATE VIEW CREATE FUNCTION ...

一、概述SQL 的执行流程一般分为四个主要的阶段，Flink 主要依赖于 Calicte 来完成这一流程： Parse：语法解析，把 SQL 语句转换成为一个抽象语法树（AST），在 Calcite 中用 SqlNode 来表示； Validate：语法校验，根据元数据信息进行验证，例如查询的表、使用的函数是否存在等，校验之后仍然是 SqlNode 构成的语法树； Optimize：查询计划优化，这里其实包含了两部分，1）首先将 SqlNode 语法树转换成关系表达式 RelNode 构成的逻辑树，2）然后使用优化器基于规则进行等价变换，例如我们比较熟悉的谓词下推、列裁剪等，经过优化器优化后得到最优的查询计划； Execute：将逻辑查询计划翻译成物理执行计划，生成对应的可执行代码，提交运行。 Flink SQL 的处理也大体遵循上述处理流程。Calcite 自身的概念较为庞杂，尤其是其内部使用的 HepPlanner 和 VolcanoPlanner 优化器更是非常复杂，但好在 Calcite 的可扩展性很强，优化器的优化规则也可以很容易地进行扩展，因此如果只是了解 Flin ...

一、概述

一、概述https://mp.weixin.qq.com/s/RKYzkYCfZ68l3EfUXS_RDA

一、概述CatalogManager 用来管理 Catalog，且可以同时管理多个 Catalog。也就是说，可以通过在一个相同 SQL 中，跨 Catalog 做查询或者关联操作。 在一个 Flink Session 当中，是可以创建多个 Catalog ，每一个 Catalog 对应于一个外部系统。用户可以在 Flink Table API 或者如果使用的是 SQL Client 的话，可以在 Yaml 文件里指定定义哪些 Catalog 。然后在 SQL Client 创建 TableEnvironment 的时候，就会把这些 Catalog 加载起来。TableEnvironment 通过CatalogManager 来管理这些不同的 Catalog 的实例。这样 SQL Client 在后续的提交 SQL 语句的过程中，就可以使用这些 Catalog 去访问外部系统的元数据了。 Flink在创建运行环境时会同时创建一个CatalogManager，用来管理不同的 Catalog 实例， https://developer.aliyun.com/article/752539 ...

一、架构设计Apache Spark 是专为大规模数据处理而设计的快速通用的计算引擎。Spark 是 UC Berkeley AMP lab(加州大学伯克利分校的AMP实验室)所开源的类 Hadoop MapReduce 的通用并行框架。 引用本站文章 Spark-理论笔记-架构设计 Joker 二、阅读环境准备 引用本站文章 Spark-源码学习-阅读环境搭建 Joker 三、集群启动Spark 采用了分布式架构的 master一slave 模型 引用本站文章 ...

一、概述UnifiedMemoryManager 是从 1.6 开始的统一内存管理模型，统一内存管理机制，与静态内存管理的区别在于存储内存和执行内存共享同一块空间，可以动态占用对方的空闲区域 二、实现2.1. 成员属性 maxHeapMemory: 最大堆内存。大小为系统可用内存与 spark.memory.fraction 属性值的乘积 onHeadpStorageRegionSize: 用于存储的堆内存大小。 numCores: CPU 内核数 2.2. 主要方法 三、内存申请3.1. 执行内存$UnifiedMemoryManager.accquireExecutionMemory()$​ 方法申请执行内存～当任务尝试从 executor 中申请 numBytes 大小的内存该方法直接向 ExecutionMemoryPool 索要所需内存: 当 ExecutionMemory 内存充足，则不会触发向 Storage 申请内存: $maybeGrowExecutionPool()$ UnifiedMemoryManager 其中最重要的优化在于动态占用机制， ...

一、概述Shuffle 是 Flink、Spark 大数据计算引擎影响计算性能的关键阶段，随着越来越多的用户选择计算存储分离的架构，将引擎部署在 K8s 集群上，而存算分离架构下计算节点 Local 磁盘不可能很大，另外 Flink、Spark 引擎还提供了根据资源量进行动态伸缩的 Adaptive Scheduler 的能力，这都要求计算节点能够将中间的 Shuffle 数据及时的卸载到外部存储服务上，以提高资源的利用效率，所以非常有必要使用独立的Shuffle 服务。 Celeborn 支持多种高效数据 Shuffle 方式，适配多种部署模式，其具备的 HA 架构、优雅下线等能力，也使得 Celeborn 自身具备弹性。 1.1. 背景1.1.1. 传统 Shuffle (External Shuffle Service) 的问题传统的 Shuffle 实现中，每个 Mapper 对 Shuffle Output 的数据，根据 Partition ID 做排序，然后把排序好的数据和索引写入本地盈。Shuffle Read 阶段，Reducer 从所有 Mapper 的 Shuf ...