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

Flink SQL解析工具类实现：从SQL到数据血缘的完整解析

news 来源：原创 2025/6/24 7:56:52

在大数据处理领域，Flink SQL作为流批统一的声明式编程接口，已成为数据处理的核心组件。本文将深入解析一个Flink SQL解析工具类的实现，该工具能够解析Flink SQL语句，提取表定义、操作关系及数据血缘信息，为数据治理、血缘分析和SQL验证提供基础能力。

工具类核心功能概述

FlinkParserUtil类实现了Flink SQL的解析功能，主要包含以下核心能力：

SQL过滤与解析：过滤自定义函数声明，解析Flink SQL中的动态表定义和操作语句
动态表解析：从CREATE TABLE语句中提取表结构、连接器类型、数据源信息
操作语句解析：解析INSERT INTO等操作语句，提取数据来源和目标表关系
血缘关系构建：分析JOIN和普通查询中的表关联关系，构建完整的数据血缘图谱
SQL验证：对SQL语句进行语法和语义验证，返回错误位置和原因

该工具类基于Apache Calcite的SQL解析能力，结合Flink SQL的语法特性，实现了对Flink SQL的完整解析流程。

核心解析流程详解

1. SQL解析入口与预处理

parserFlinkSql方法是整个解析流程的入口，负责协调各个解析步骤：

public static Set<FlinkTable> parserFlinkSql(String flinkSql) throws SqlParseException {// 过滤自定义函数声明，避免解析报错String[] split = flinkSql.split(";\n");String sql = Arrays.stream(split).filter(v -> !v.trim().startsWith(CUSTOM_FUNCTION)).collect(Collectors.joining(";"));List<SqlCreateTable> dynamicTables = new ArrayList<>();List<RichSqlInsert> operationTables = new ArrayList<>();// 构建Flink SQL解析器SqlParser parser = buildSqlParser(sql);List<SqlNode> sqlNodeList = parser.parseStmtList().getList();// 分类解析SQL节点sqlNodeList.forEach(v -> {if (v instanceof SqlCreateTable) {dynamicTables.add((SqlCreateTable) v);} else if (v instanceof RichSqlInsert) {operationTables.add((RichSqlInsert) v);}});// 解析动态表和操作表，构建血缘关系Map<String, FlinkTable> dynamicTableMap = parseDynamicTable(dynamicTables);Set<FlinkTable> operationTableMap = parseOperation(operationTables);return parseFlinkBlood(dynamicTableMap, operationTableMap);
}

解析流程首先过滤掉自定义函数声明（避免解析报错），然后使用Flink定制的SQL解析器将SQL语句转换为抽象语法树（AST），最后分类处理CREATE TABLE和INSERT INTO等语句。

2. 动态表解析与元数据提取

parseDynamicTable方法负责解析CREATE TABLE语句，提取表结构和连接信息：

private static Map<String, FlinkTable> parseDynamicTable(List<SqlCreateTable> dynamicTables) {Map<String, FlinkTable> dynamicTableMap = new HashMap<>();dynamicTables.forEach(v -> {FlinkTable tbl = new FlinkTable();String flinkTableName = v.getTableName().toString();tbl.setFlinkTableName(flinkTableName);// 提取表结构字段List<SqlNode> list = v.getColumnList().getList();Set<String> columns = list.stream().map(m -> {SqlTableColumn column = (SqlTableColumn) m;return String.valueOf(column.getName());}).collect(Collectors.toSet());tbl.setColumnList(columns);// 提取表属性（连接器、主题、URL等）List<SqlNode> propertyList = v.getPropertyList().getList();for (SqlNode sqlNode : propertyList) {SqlTableOption option = (SqlTableOption) sqlNode;String optionKey = option.getKey().toString();String value = option.getValue().toString().replaceAll("'", "");switch (optionKey) {case TOPIC:tbl.setSourceTableName(value);break;case TABLE_NAME:tbl.setSourceTableName(value);break;case CONNECTOR:tbl.setConnectorName(value);break;// 其他属性处理case URL:tbl.setUrl(value);break;case USERNAME:tbl.setUsername(value);break;case PASSWORD:tbl.setPassword(value);break;case SERVERS:tbl.setServers(value);break;}}dynamicTableMap.put(flinkTableName, tbl);});return dynamicTableMap;
}

该方法从CREATE TABLE语句中提取表名、字段列表和表属性（如connector、topic、servers等），封装为FlinkTable对象，为后续血缘分析提供基础元数据。

3. 操作语句解析与血缘构建

parseOperation方法解析INSERT INTO等操作语句，提取数据来源：

private static Set<FlinkTable> parseOperation(List<RichSqlInsert> operationTables) {Set<FlinkTable> tableSet = new HashSet<>();operationTables.forEach(v -> {FlinkTable tbl = new FlinkTable();tbl.setFlinkTableName(String.valueOf(v.getTargetTable()));SqlSelect source = (SqlSelect) v.getSource();Map<String, Set<String>> sourceMap = new HashMap<>();SqlNode sourceFrom = source.getFrom();// 处理JOIN操作和普通查询if (sourceFrom instanceof SqlJoin) {sourceMap.putAll(parseJoinOperator(source));} else if (sourceFrom instanceof SqlIdentifier) {sourceMap.putAll(parseCommonOperator(source));}// 构建来源表集合Set<FlinkTable> sourceSet = sourceMap.keySet().stream().map(key -> {FlinkTable sourceTable = new FlinkTable();sourceTable.setFlinkTableName(key);sourceTable.setColumnList(sourceMap.get(key));return sourceTable;}).collect(Collectors.toSet());tbl.setSourceSet(sourceSet);tableSet.add(tbl);});return tableSet;
}

对于不同类型的查询（JOIN或普通查询），工具类使用不同的解析策略：

JOIN操作解析

parseJoinOperator方法专门处理JOIN操作，提取多表关联关系：

public static Map<String, Set<String>> parseJoinOperator(SqlSelect sqlNode) {Map<String, Set<String>> sourceMap = new HashMap<>();SqlJoin join = (SqlJoin) sqlNode.getFrom();SqlBasicCall left = (SqlBasicCall) join.getLeft();SqlBasicCall right = (SqlBasicCall) join.getRight();// 解析JOIN左右表关系SqlNode[] leftOperands = left.getOperands();Map<String, String> relateMap = new HashMap<>();if (leftOperands.length >= 1) {relateMap.put(String.valueOf(leftOperands[DEFAULT_INDEX + 1]), String.valueOf(leftOperands[DEFAULT_INDEX]));} else {relateMap.put(String.valueOf(leftOperands[DEFAULT_INDEX]), String.valueOf(leftOperands[DEFAULT_INDEX]));}SqlNode[] rightOperands = right.getOperands();if (rightOperands.length >= 1) {String[] relDynamicTable = String.valueOf(rightOperands[DEFAULT_INDEX]).trim().split(" ");relateMap.put(String.valueOf(rightOperands[DEFAULT_INDEX + 1]), String.valueOf(relDynamicTable[DEFAULT_INDEX]).replaceAll(QUOTE, ""));} else {relateMap.put(String.valueOf(leftOperands[DEFAULT_INDEX]), String.valueOf(leftOperands[DEFAULT_INDEX]));}// 解析SELECT字段对应的表和列List<SqlNode> list = sqlNode.getSelectList().getList();list.forEach(v -> {SqlBasicCall sqlBasicCall = (SqlBasicCall) v;String operand = Arrays.stream(sqlBasicCall.getOperands()).findFirst().get().toString();String[] split = operand.split(SEPARATOR);String key = relateMap.get(split[DEFAULT_INDEX]);String value = split[DEFAULT_INDEX + 1];if (sourceMap.containsKey(key)) {sourceMap.get(key).add(value);} else {sourceMap.put(key, new HashSet<>(Collections.singletonList(value)));}});return sourceMap;
}

普通查询解析

parseCommonOperator方法处理普通查询语句，提取单表数据来源：

private static Map<String, Set<String>> parseCommonOperator(SqlSelect source) {Map<String, Set<String>> sourceMap = new HashMap<>();Map<String, String> relateMap = new HashMap<>();// 提取FROM子句中的表名SqlIdentifier from = (SqlIdentifier) source.getFrom();String key = from.getSimple();relateMap.put(key, key);// 提取SELECT字段List<SqlNode> list = source.getSelectList().getList();Set<String> columnSet = new HashSet<>();list.forEach(v -> {if (v instanceof SqlIdentifier) {SqlIdentifier identifier = (SqlIdentifier) v;columnSet.add(identifier.getSimple());} else if (v instanceof SqlBasicCall) {SqlBasicCall call = (SqlBasicCall) v;SqlNode[] operands = call.getOperands();if (operands.length <= 0) {SqlIdentifier sqlIdentifier = (SqlIdentifier) operands[0];columnSet.add(sqlIdentifier.getSimple());} else {SqlBasicCall sqlNode = (SqlBasicCall) Arrays.stream(operands).filter(f -> f instanceof SqlBasicCall).findFirst().get();SqlNode operand = sqlNode.getOperands()[0];if (operand instanceof SqlIdentifier) {columnSet.add(((SqlIdentifier) operand).getSimple());} else if (operand instanceof SqlBasicCall) {SqlBasicCall basicCall = (SqlBasicCall) sqlNode.getOperands()[0];SqlIdentifier identifier = (SqlIdentifier) basicCall.getOperands()[0];columnSet.add(identifier.getSimple());}}}});sourceMap.put(key, columnSet);return sourceMap;
}

4. 血缘关系整合

parseFlinkBlood方法整合动态表和操作表信息，构建完整的血缘关系：

private static Set<FlinkTable> parseFlinkBlood(Map<String, FlinkTable> dynamicTableMap,Set<FlinkTable> operationTableSet) {return operationTableSet.stream().map(tbl -> {String flinkTableName = tbl.getFlinkTableName();FlinkTable table = dynamicTableMap.get(flinkTableName);// 填充目标表的来源表和连接器信息tbl.setSourceTableName(table.sourceTableName.replaceAll("'", ""));tbl.setColumnList(table.getColumnList());tbl.setConnectorName(table.getConnectorName());// 递归处理来源表的血缘关系Set<FlinkTable> tableSet = tbl.getSourceSet().stream().map(v -> {String sourceKey = tbl.getFlinkTableName();FlinkTable source = dynamicTableMap.get(sourceKey);v.setSourceTableName(source.sourceTableName);v.setConnectorName(source.getConnectorName());return v;}).collect(Collectors.toSet());tbl.setSourceSet(tableSet);return tbl;}).collect(Collectors.toSet());
}

数据结构设计

FlinkTable类作为核心数据结构，存储解析得到的表信息：

public static class FlinkTable {private String flinkTableName;         // Flink中定义的表名private String sourceTableName;        // 实际数据源表名private Set<String> columnList;        // 表结构字段private String connectorName;          // 连接器类型private Set<FlinkTable> sourceSet;     // 来源表集合private String url;                    // 连接URLprivate String username;               // 用户名private String password;               // 密码private String servers;                // 服务器地址//  getter和setter方法//  toString方法
}

该结构完整存储了表定义、连接信息和血缘关系，为后续数据治理和血缘分析提供了丰富的元数据。

SQL验证功能

工具类还提供了SQL验证功能，能够检测SQL语句中的语法和语义错误：

public static Map<String, Position> validateSql(String validateSql) throws SqlParseException {Map<String, Position> validateMap = new HashMap<>();SqlNode sqlNode = buildSqlParser(validateSql).parseStmt();// 使用Calcite的验证器进行语义验证SqlValidator validator = new SqlAdvisorValidator(null, null, null, null);ListScope scope = new ListScope(null) {@Overridepublic SqlNode getNode() {return null;}};sqlNode.validate(validator, scope);// 模拟验证结果（实际应用中可根据验证器错误信息填充）Position position = new Position();position.setEnd(10);position.setStart(0);position.setMsg("column name not find exist table");validateMap.put("userName", position);return validateMap;
}static class Position {private Integer start;     // 错误开始位置private Integer end;       // 错误结束位置private String msg;        // 错误信息private Integer line;      // 错误行号// getter和setter方法
}