通过Spark SQL实现行级过滤和列脱敏

版本要求

Spark能力依赖AnalyticDB MySQL升级到特定内核版本。低于3.2.8版本的集群需要先进行集群升级。

权限要求

• 策略管理DDL（CREATE、UPDATE、DROP、BIND、UNBIND、SHOW）必须由管理员账号执行。普通用户执行策略DDL会收到AccessDenied错误。

• 普通用户无需任何额外授权即可被策略约束：策略绑定到谁，谁就在查询时自动受约束。

用户身份

• 策略绑定与生效判定基于ADB用户名（即登录账号）。BIND ... TO '<user>'中的用户名必须与目标用户的ADB登录账号完全一致。

• 查询时系统自动从登录态解析当前用户身份，再匹配该用户已绑定的策略。身份解析失败时查询会被中止（fail-closed），不会在无保护状态下放行。

连接方式

通过JDBC（Beeline、DBeaver或应用程序）连接Spark交互式资源组后即可使用，示例如下：

jdbc:hive2://<spark-endpoint>:10000/<database>?resource_group=<资源组名>

适用场景

• 多租户数据隔离：每个租户的分析师只能看到本租户的数据行（如tenant_id = 'T001'）。

• 区域或组织级数据可见性：区域运营只能看到本区域数据（如region = 'APAC'）。

• 敏感记录保护：普通员工不能看到高管、薪酬超阈值等敏感记录（如salary < 50000）。

• 合规审计要求：法规要求特定人群只能访问脱敏子集的数据行。

SQL语句参考

使用示例

以下示例基于演示表sec_e2e.employees：

-- 管理员准备演示数据
CREATE DATABASE IF NOT EXISTS sec_e2e;
CREATE TABLE IF NOT EXISTS sec_e2e.employees (
  id INT, name STRING, region STRING, department STRING, salary DOUBLE, phone STRING
) USING parquet;

INSERT INTO sec_e2e.employees VALUES
  (1, 'Alice', 'APAC', 'HR',          75000.0,  '13812341234'),
  (2, 'Bob',   'EU',   'Engineering', 95000.0,  '13956785678'),
  (3, 'Carol', 'US',   'Finance',     85000.0,  '13790129012'),
  (4, 'David', 'APAC', 'Engineering', 105000.0, '13634563456'),
  (5, 'Eve',   'EU',   'HR',          70000.0,  '13578907890');

场景：APAC区域分析师global_read只能看到APAC区域的员工数据。

-- ① 管理员创建行过滤策略
CREATE ROW FILTER POLICY rf_apac ON sec_e2e.employees USING (region = 'APAC');

-- ② 管理员将策略绑定到目标用户
BIND ROW FILTER POLICY rf_apac TO 'global_read';

-- ③ 管理员核对策略与绑定关系
SHOW ROW FILTER POLICIES IN sec_e2e;
-- PolicyName | Database | Table     | Condition
-- rf_apac    | sec_e2e  | employees | region = 'APAC'

SHOW ROW FILTER BINDINGS FOR POLICY rf_apac;
-- PolicyName | Database | Table     | Condition        | User
-- rf_apac    | sec_e2e  | employees | region = 'APAC'  | global_read

被绑定用户global_read查询验证（用户侧无需改写任何SQL）：

SELECT * FROM sec_e2e.employees;
-- 仅返回 region = 'APAC' 的两行（id=1 Alice、id=4 David），
-- EU / US 的行对该用户完全不可见

过滤条件支持任意合法的SQL谓词表达式，包括函数与UDF：

-- 多条件组合
CREATE ROW FILTER POLICY rf_dept ON sec_e2e.employees
  USING (department = 'HR' AND salary > 71000.0);

-- 使用自定义函数（UDF）作为过滤条件
CREATE FUNCTION IF NOT EXISTS is_hr_with_valid_int
  AS 'com.aliyun.adb.udf.IsHrWithValidInt' USING JAR 'oss://<bucket>/hr-udf-1.0.0.jar';

CREATE ROW FILTER POLICY rf_hr_high_salary ON sec_e2e.employees
  USING (is_hr_with_valid_int(department, CAST(phone AS BIGINT)) AND salary > 71000.0);

策略维护：

-- 更新过滤条件（对所有已绑定用户立即生效）
UPDATE ROW FILTER POLICY rf_apac USING (region = 'EU');

-- 解绑用户（该用户立即恢复全量可见）
UNBIND ROW FILTER POLICY rf_apac FROM 'global_read';

-- 删除策略（级联清除所有绑定关系）
DROP ROW FILTER POLICY rf_apac;

支持幂等创建，适合脚本化和自动化管理：

-- 策略已存在时静默跳过，不报错、不覆盖原条件
CREATE ROW FILTER POLICY IF NOT EXISTS rf_apac ON sec_e2e.employees USING (region = 'APAC');

生效范围与防绕过

行过滤在查询分析阶段自动注入，对用户透明且无法通过SQL写法绕过：

• 子查询：SELECT * FROM (SELECT * FROM sec_e2e.employees) t，内层扫描同样被过滤。

• CTE（WITH子句）：WITH cte AS (SELECT * FROM sec_e2e.employees) SELECT * FROM cte，同样被过滤。

• 视图（含临时视图）：创建临时视图后查询，同样被过滤。

• JOIN：过滤条件只作用于绑定了策略的表，未绑定策略的关联表不受影响。

写入保护（禁写）

开启行过滤（rowFilter.enabled=true）后，禁写保护默认同时开启（blockWrites.enabled=true）：

• 当前用户在某表上绑定了行过滤策略时，禁止其对该表执行任何数据修改操作，包括INSERT（INTO/OVERWRITE）、UPDATE、DELETE、MERGE、TRUNCATE TABLE、LOAD DATA、REPLACE TABLE和ALTER TABLE ... DROP PARTITION。

• 违规写入会报错并中止。

不受禁写影响的场景：

• 未绑定策略的用户（含管理员）对同一张表的写入照常。

• CREATE TABLE ... AS SELECT（CTAS）写入新表放行（新表不存在既有策略绑定）。

• INSERT INTO 无策略表 SELECT ... FROM 有策略表放行，写入目标表无策略，读取源表时行过滤照常生效。

限制说明

• 默认关闭：spark.adb.security.rowFilter.enabled默认为false，未开启时策略不生效、查询返回明文（fail-open）。开关为静态配置，需在资源组或作业启动配置中设置。

• 仅覆盖内置Catalog：行过滤仅对spark_catalog（ADB内置元数据目录）下的表生效。通过外部Catalog（如external-hive、paimon等V2 Catalog）访问的表不在策略范围内。

• 同一用户在同一张表上只允许绑定一个行过滤策略：重复绑定第二个策略时，BIND语句直接报错。如需切换策略，先UNBIND旧策略再BIND新策略。

• 一张表可以创建多个行过滤策略（策略名需唯一），用于绑定给不同用户群，但同一用户只能命中其中一条。

• 过滤条件在创建时做语法预校验，但条件中引用的列或函数在查询时才真正解析。若策略创建后表结构变更导致条件无法解析，被绑定用户的查询将报错中止（fail-closed），而不是放行。

• 策略变更的生效时间：执行DDL的会话立即生效；其他已运行的Spark应用因存在策略缓存（默认5分钟TTL），最长延迟5分钟生效。

• 跨引擎冲突：若同一张表在其他引擎（XIHE）上已存在行过滤策略，Spark侧创建策略会报错，且该冲突无法在Spark侧自行解除，需联系管理员在原引擎侧处理。

• 元数据服务未配置时功能降级：环境未配置元数据服务端点时，行过滤自动降级为无操作（普通查询不受影响，但安全保护不生效）。

适用场景

• 个人隐私信息（PII）保护：手机号、身份证号、邮箱等对普通分析人员脱敏展示（如138****1234）。

• 薪酬等敏感字段保护：普通员工看不到真实薪资，只能看到哈希值或掩码。

• 合规要求：法规要求敏感字段在非授权场景下不可见明文。

• 数据共享或外包分析：外部协作账号可以使用数据做分析，但敏感列一律脱敏。

SQL语句参考

列脱敏DDL与行过滤同构，将ROW FILTER替换为MASKING，并额外指定目标列：

使用示例

沿用行级过滤示例中的sec_e2e.employees表，目标用户为global_read。

部分脱敏（保留前3位与后4位的手机号掩码）：

CREATE MASKING POLICY cm_phone ON sec_e2e.employees COLUMN phone
  USING (concat(substr(phone, 1, 3), '****', substr(phone, 8, 4)));

BIND MASKING POLICY cm_phone TO 'global_read';

global_read查询效果：

SELECT id, name, phone FROM sec_e2e.employees;
-- id | name  | phone
-- 1  | Alice | 138****1234
-- 2  | Bob   | 139****5678
-- ...（phone 列全部按掩码规则展示，明文不可见）

哈希脱敏（不可逆摘要，可用于关联分析但不泄露原值）：

CREATE MASKING POLICY cm_salary ON sec_e2e.employees COLUMN salary
  USING (md5(cast(salary AS STRING)));

BIND MASKING POLICY cm_salary TO 'global_read';
-- global_read查询salary列返回32位十六进制串，如'5f4dcc3b5aa765d61d8327deb882****'

全脱敏（常量掩码）：

CREATE MASKING POLICY cm_name ON sec_e2e.employees COLUMN name
  USING ('********');

BIND MASKING POLICY cm_name TO 'global_read';
-- global_read查询name列全部显示'********'

NULL脱敏（完全隐藏列值）：

CREATE MASKING POLICY cm_region ON sec_e2e.employees COLUMN region
  USING (CAST(NULL AS STRING));

BIND MASKING POLICY cm_region TO 'global_read';
-- global_read查询region列全部返回NULL

多列同时脱敏：

一张表可创建多条脱敏策略分别作用于不同列，未设置脱敏的列保持原值。

SELECT * FROM sec_e2e.employees WHERE id = 1;
-- global_read 视角：
-- id=1, name='********', region=NULL, department='HR'（原值）,
-- salary=<32位md5>, phone='138****1234'

策略维护：

-- 更新脱敏表达式
UPDATE MASKING POLICY cm_phone USING ('********');

-- 查看某用户绑定的全部脱敏策略
SHOW MASKING BINDINGS FOR USER 'global_read';

-- 解绑后该用户立即恢复看到原值
UNBIND MASKING POLICY cm_phone FROM 'global_read';

-- 删除策略（级联清除绑定）
DROP MASKING POLICY cm_phone;

生效范围与防绕过

与行过滤一致，列脱敏在查询分析阶段自动注入，子查询、CTE、视图（含临时视图）均无法绕过；JOIN场景下仅对绑定策略的表的对应列脱敏。

限制说明

• 默认关闭：spark.adb.security.columnMask.enabled默认为false，未开启时策略不生效、查询返回明文（fail-open）。

• 仅覆盖内置Catalog：仅对spark_catalog下的表生效，外部Catalog表不在策略范围内。

• 脱敏表达式仅支持纯标量表达式，以下写法会被拒绝：

  • 聚合函数（如sum(col)、max(col)）。

  • 窗口函数或窗口表达式（如row_number() OVER (...)）。

  • DISTINCT、FILTER修饰的函数调用。

  • 三段式以上的函数名（仅支持fn或db.fn两种形式）。

  • 引用当前表以外的列（表达式中引用的列必须属于策略所在表）。

• 同一用户在同一张表的同一列上，查询时只允许命中一条脱敏策略。若因历史原因同一（表, 列, 用户）存在多条绑定，查询将报错中止（fail-closed），需管理员清理至恰好保留一条。

• 脱敏后列的数据类型可能变化：输出列名保持不变，但类型由脱敏表达式决定。下游报表或应用若依赖原始列类型，需提前评估。

• 列脱敏只影响读取，不拦截写入：被绑定用户的INSERT等写操作照常执行（写入的是原值），但其读回的数据按策略脱敏。若需要绑定即禁写的强保护，请使用行过滤。

• 表达式在创建时做语法预校验，表达式中引用的列或函数在查询时才真正解析绑定。若创建后表结构变更导致表达式无法解析，被绑定用户的查询将报错中止（fail-closed）。

• 策略变更的生效时间：执行DDL的会话立即生效；其他已运行的Spark应用最长延迟5分钟（策略缓存TTL）生效。

• 跨引擎冲突：若同一张表的同一列在其他引擎（XIHE）上已存在脱敏策略，Spark侧创建策略会报错，需联系管理员在原引擎侧处理。