当前位置： 首页 > backend >正文

Spark SQL架构及高级用法

backend 2025/7/6 9:54:54

Spark SQL 架构概述

架构核心组件

  • API层（用户接口）

    • 输入方式：SQL查询；DataFrame/Dataset API。
    • 统一性： 所有接口最终转换为逻辑计划树（Logical Plan），进入优化流程。

  • 编译器层（Catalyst 优化器）

    • 核心引擎： 基于规则的优化器（Rule-Based Optimizer, RBO）与成本优化器（Cost-Based Optimizer, CBO）。

    • 处理流程：

      阶段输入输出关键动作
      解析SQL/API 操作未解析逻辑计划构建语法树（AST），校验语法正确性
      分析未解析逻辑计划解析后逻辑计划绑定元数据（表/列名、数据类型）、解析函数、检查语义正确性
      优化解析后逻辑计划优化后逻辑计划应用优化规则（如谓词下推、列剪裁、常量折叠、连接重排序）

    • 优化规则示例：

      Predicate Pushdown（谓词下推）：将过滤条件推至数据源层，减少 I/O。

      Column Pruning（列裁剪）：仅读取查询涉及的列，减少数据传输。

  • 执行计划层（Planner）

    • 物理计划生成：将优化后的逻辑计划转换为物理计划（Physical Plan）。

    • 策略匹配： 根据数据分布、资源情况选择最优执行策略（如 BroadcastHashJoin vs SortMergeJoin）。

    • 物理优化：

      全阶段代码生成（Whole-Stage Codegen）：将多个操作合并为单个 JVM 函数，减少虚函数调用开销。

      谓词下推至数据源：支持 Parquet/ORC 等格式的过滤条件下推。

  • 执行引擎层（Tungsten + Spark Core）

    • Tungsten 引擎：

      堆外内存管理：避免 JVM GC 开销，直接操作二进制数据。

      向量化计算：按列处理数据，提升 CPU 缓存命中率。

    • 分布式执行：

      物理计划转为 RDD DAG → 分解为 Stage → 调度 Task 到 Executor 并行执行。

      利用 Spark Core 的血缘（Lineage）、内存管理、Shuffle 服务。

关键性能技术
  • Catalyst 优化器
    • 动态优化： 在逻辑计划阶段应用启发式规则，减少冗余计算。
    • 自适应查询（AQE, Spark 3.0+）：运行时根据 Shuffle 数据量动态调整 Join 策略、分区数。
  • Tungsten 执行引擎
    • 内存效率： 紧凑二进制格式存储数据，减少内存占用 50%+。
    • 代码生成： 将查询编译为字节码，性能接近手写代码。
  • 统一数据源接入
    • Data Source API V2： 支持扩展自定义数据源（如 Kafka、Cassandra），并实现下推优化。
graph LRA[SQL 查询] --> B(Parser：生成语法树)B --> C(Analyzer：解析元数据)C --> D(Optimizer：应用优化规则)D --> E(Planner：生成物理计划)E --> F(Tungsten：代码生成+内存优化)F --> G(Spark Core：分布式执行)

Spark SQL高级语法

复杂数据类型

  • 数组 (ARRAY)：同类型元素的有序集合（索引从0开始）。

    • size()：数组长度。
    • explode()：展开数组为多行。
    • array_contains(arr, value)：检查元素是否存在。
    • transform(arr, x -> x * 2)：对每个元素应用Lambda函数。

  • 映射(MAP<K,V>)：键值对集合（键唯一）。

    • element_at(map, key)：按键取值。

    • map_keys()/map_values()：获取所有键/值。

    • map_concat(map1, map2)：合并两个Map。

  • 结构体 (STRUCT<field1:T1, ...>)：包含多个字段的复合类型（类似JSON对象）。

高级聚合与分组

  • GROUPING SETS：自定义聚合维度组合，无关字段用NULL值填充。

    SELECT city, department, SUM(salary) AS total_salary
FROM employees
GROUP BY GROUPING SETS ((city, department), (city), ()                  
)

  • ROLLUP：层级聚合

    SELECT country, province, city,COUNT(*) AS count
FROM locations
GROUP BY ROLLUP(country, province, city)
-- SQL结果会显示（country, province, city）、（country, province）、（country）、（）的聚合结果

  • CUBE：所有维度聚合

    SELECT year, product, SUM(revenue) 
FROM sales
GROUP BY CUBE(year, product)
-- SQL结果会显示（year, profucr）、（year）、（product）、（）的聚合结果

  • 聚合过滤（FILTER 子句）：对特定条件聚合 (比WHERE更高效)

    SELECT department,SUM(salary) FILTER (WHERE age > 30) AS senior_salary,AVG(salary) FILTER (WHERE gender = 'F') AS female_avg
FROM employees
GROUP BY department
窗口函数

  • 核心结构

    SELECTRANK() OVER (PARTITION BY dim ORDER BY metric DESCROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW -- 窗口范围) AS rank
FROM table

  • 聚合函数：**SUM()、AVG()、COUNT()、MAX()、MIN()**等

  • 排名函数：

    • ROW_NUMBER()：为窗口内的每一行分配一个唯一的序号，序号连续且不重复。
    • RANK()：排名函数，允许有并列的名次，名次后面会出现空位。
    • ENSE_RANK()：排名函数，允许有并列的名次，名次后面不会空出位置，即序号连续。

  • 分组窗口函数：

    • NTILE()：将窗口内的行分为指定数量的组，每组的行数尽可能相等。

  • 分布窗口函数

    • PERCENT_RANK()：计算每一行的相对排名，返回一个介于0到1之间的值，表示当前行在分区中的排名百分比。
    • CUME_DIST()：计算小于或等于当前行的行数占窗口总行数的比例。

  • 取值窗口函数

    • LAG()：访问当前行之前的第n行数据。
    • LEAD()：访问当前行之后的第n行数据。
    • FIRST_VALUE()：获取窗口内第一行的值。
    • LAST_VALUE()：获取窗口内最后一行的值。
    • NTH_VALUE()：获取窗口内第n行的值，如果存在多行则返回第一个。

  • 窗口范围

    • UNBOUNDED PRECEDING：从分区中的第一行开始（前面所有行）。
    • CURRENT ROW：包括当前行。
    • N PRECEDING：从当前行之前的第 nN行开始。
    • N FOLLOWING：包括当前行之后第 N 行。
    • UNBOUNDED FOLLOWING：到分区中的最后一行结束（后面所有行）。
Spark SQL内置函数

  • 聚合函数 (Aggregate Functions)

    函数返回值说明
    approx_count_distinctLong近似去重计数 (rsd=相对误差)
    collect_listArray收集值到数组 (保留重复)
    collect_setArray收集值到集合 (去重)
    corrDouble相关系数 (-1~1)
    covar_pop/covar_sampDouble总体/样本协方差
    kurtosisDouble峰度
    skewnessDouble偏度
    percentile_approxDouble近似百分位数

  • 数组函数 (Array Functions)

    函数说明
    array(e1, e2, …)转换为数组
    array_contains(arr, val)数组包含
    array_distinct数组去重
    array_position(arr, val)数组索引值
    size数组大小

  • Map 函数 (Map Functions)

    函数说明
    map(k1,v1, k2,v2)转化为map
    element_at(map, key)根据键获取值
    map_keys/map_values获取键列表、值列表
    map_entries获取map entry

  • 日期时间函数 (Datetime Functions)

    函数说明示例
    date_add/date_sub日期加减date_add('2025-07-01', 7) → 2025-07-08
    datediff日期差datediff('2025-06-01','2025-06-30') → 30
    date_format格式化date_format(ts, 'yyyy-MM') → "2025-07"
    trunc截断日期trunc('2025-07-01', 'MONTH') → 2025-07-01
    window时间窗口流处理中按时间聚合

  • JSON 函数 (JSON Functions)

    函数说明示例
    get_json_objectJSON路径取值get_json_object('{"a":1}', '$.a') → 1
    json_tuple多字段提取json_tuple('{"name":"Bob"}', 'name') → Bob
    from_json解析为结构体from_json('{"id":1}', 'id INT')
    to_json结构体转JSONto_json(struct('Tom' AS name)) → '{"name":"Tom"}'
    schema_of_json推断Schemaschema_of_json('[{"a":1}]') → ARRAY<STRUCT<a:INT>>

  • 字符串函数 (String Functions)

    函数说明示例
    concat_ws字符串拼接joinconcat_ws('-','2025','07')→"2025-07"
    split字符串解析转数组split('a,b,c', ',')→["a","b","c"]
    regexp_extract正则表达式regexp_extract('id=100','id=(\\d+)',1)→"100"
    translate字符串子串转化translate('hello','el','ip')→"hippo"
    parse_url解析URLparse_url('http://a.com?q=spark','QUERY')→"q=spark"
http://www.xdnf.cn/news/14884.html

相关文章：

  • Zama 机密区块链协议Litepaper
  • Android Input 系列专题【事件的读取与分发】
  • 408第三季part2 - 计算机网络 - 计算机网络分层结构
  • MySQL——9、事务管理
  • Linux安装java后没法运行
  • StarRocks × Tableau 连接器完整使用指南 | 高效数据分析从连接开始
  • 智能监控算法助力工厂高温高效管理
  • Windows内核并发优化
  • 使用SSH隧道连接远程主机
  • H3C WA6322 FitAP切换为云AP或
  • 不依赖框架，如何用 JS 实现一个完整的前端路由系统
  • 快速掌握Python编程基础
  • Day51 复习日-模型改进
  • 免费PDF处理软件，支持多种操作
  • 深度学习笔记29-RNN实现阿尔茨海默病诊断(Pytorch)
  • 因果语言模型、自回归语言模型、仅解码器语言模型都是同一类模型
  • VS CodeC51 单片机开发环境搭建
  • Django+DRF 实战：从异常捕获到自定义错误信息
  • 从入门到精通：数据库全攻略
  • Jenkins-Publish HTML reports插件
  • 技术与情感交织的一生 （九）
  • Python 的内置函数 reversed
  • WPS中配置MathType教程
  • 数学中微分与导数的理解和区别
  • Linux基本命令篇 —— uname命令
  • 本地部署Dify并结合ollama大语言模型工具搭建自己的AI知识库
  • 垂起固定翼无人机介绍
  • 二叉树题解——将有序数组转换为二叉搜索树【LeetCode】传统解法
  • Compose 高级用法详解——AI教你学Docker
  • 焊接机器人结构设计cad【16张】三维图＋设计说明书＋绛重