如何从Python脚本中提取所有SQL查询涉及的表名？

方案一：正则匹配（快速实现，适合简单场景） #

直接通过正则表达式从Python脚本中提取spark.sql()里的SQL语句，再从中匹配表名，步骤如下：

1. 先匹配所有spark.sql("...")或spark.sql('...')形式的代码段，抽取出内部的SQL字符串。
2. 针对每个SQL字符串，匹配FROM、JOIN、INTO等关键字后的表名（支持schema.table格式）。

代码示例：

import re

# 读取目标Python脚本
with open("your_script.py", "r", encoding="utf-8") as f:
    script_content = f.read()

# 提取所有spark.sql中的SQL语句
sql_extract_pattern = re.compile(r'spark\.sql\(["\'](.*?)["\']\)', re.DOTALL | re.IGNORECASE)
sql_statements = sql_extract_pattern.findall(script_content)

# 提取SQL中的表名
table_extract_pattern = re.compile(r'(?:FROM|JOIN|INTO)\s+([\w.]+)', re.IGNORECASE)
all_tables = set()
for sql in sql_statements:
    matched_tables = table_extract_pattern.findall(sql)
    all_tables.update(matched_tables)

print("提取到的表名：", all_tables)

局限：正则无法处理复杂SQL场景，比如SQL中嵌套子查询、注释里包含关键字、字符串常量中出现FROM等情况，会出现误匹配或漏匹配。

方案二：SQL解析器（更可靠，适配复杂场景） #

使用专门的SQL解析库或Spark自带的语法解析器，解析SQL的抽象语法树（AST）来精准提取表名，这是更优的方案。

子方案1：用sqlparse库（无需Spark环境）

sqlparse是Python的SQL解析库，能处理绝大多数标准SQL语法，遍历AST节点提取表名。

代码示例：

import re
import sqlparse
from sqlparse.sql import IdentifierList, Identifier
from sqlparse.tokens import Token

def get_tables_from_sql(sql):
    tables = set()
    parsed_sql = sqlparse.parse(sql)[0]
    
    # 遍历解析后的SQL节点
    for token in parsed_sql.tokens:
        # 处理多表情况（比如JOIN多个表）
        if isinstance(token, IdentifierList):
            for item in token.get_identifiers():
                if isinstance(item, Identifier):
                    tables.add(item.get_real_name())
        # 处理单个表的情况
        elif isinstance(token, Identifier):
            parent_token = token.parent
            if parent_token and parent_token.ttype == Token.Keyword and parent_token.value.upper() in ('FROM', 'JOIN'):
                tables.add(token.get_real_name())
    return tables

# 读取脚本并提取SQL语句
with open("your_script.py", "r", encoding="utf-8") as f:
    script_content = f.read()

sql_extract_pattern = re.compile(r'spark\.sql\(["\'](.*?)["\']\)', re.DOTALL | re.IGNORECASE)
sql_statements = sql_extract_pattern.findall(script_content)

all_tables = set()
for sql in sql_statements:
    tables = get_tables_from_sql(sql)
    all_tables.update(tables)

print("提取到的表名：", all_tables)

子方案2：用Spark自带解析器（最准确，需Spark环境）

如果你的环境已经部署Spark，可以直接用Spark的SQL解析器，它完全适配Spark SQL的语法特性（包括临时视图、分区表、自定义函数等），能精准提取所有涉及的表。

代码示例：

import re
from pyspark.sql import SparkSession

# 初始化SparkSession
spark = SparkSession.builder.appName("TableExtractor").getOrCreate()

def extract_spark_sql_tables(sql):
    tables = set()
    # 解析SQL生成逻辑计划
    logical_plan = spark.sqlParser.parsePlan(sql)
    
    # 递归遍历逻辑计划，找到表扫描节点
    def traverse_plan(node):
        # 匹配表扫描节点（包含tableName属性）
        if hasattr(node, 'tableName'):
            tables.add(node.tableName)
        # 递归遍历子节点
        elif hasattr(node, 'children'):
            for child in node.children:
                traverse_plan(child)
    
    traverse_plan(logical_plan)
    return tables

# 读取脚本并提取SQL语句
with open("your_script.py", "r", encoding="utf-8") as f:
    script_content = f.read()

sql_extract_pattern = re.compile(r'spark\.sql\(["\'](.*?)["\']\)', re.DOTALL | re.IGNORECASE)
sql_statements = sql_extract_pattern.findall(script_content)

all_tables = set()
for sql in sql_statements:
    tables = extract_spark_sql_tables(sql)
    all_tables.update(tables)

print("提取到的表名：", all_tables)
# 停止SparkSession
spark.stop()

方案选择建议 #

• 若你的SQL都是简单结构（无嵌套子查询、无复杂注释），用正则匹配快速实现即可。
• 若涉及复杂SQL场景，优先选择sqlparse（无需依赖Spark）或Spark自带解析器（最准确）。

内容的提问来源于stack exchange，提问作者kiruba