pyspark 本身不提供将 dataframe 链式操作（如 select、filter）直接转为可执行 sql 字符串的内置方法，但可通过解析其底层逻辑计划（logicalplan）手动提取 select、from 和 where 等关键成分，构造近似等价的 sql 表达式。
在 PySpark 中，DataFrame 是惰性求值的，其执行逻辑由 Catalyst 优化器管理，所有操作最终都会编译为一个逻辑执行计划（LogicalPlan）。虽然 Spark 并未开放稳定、官方支持的“DataFrame → SQL”反向生成 API（因逻辑计划是内部表示，且可能包含优化后不可逆的结构），但我们可以通过访问私有 Java 对象（_jdf.queryExecution().logical()）获取计划字符串，并进行启发式解析，实现对简单链式操作（如 select + where）的 SQL 还原。
以下是一个轻量级、可扩展的参考实现：
from pyspark.sql import functions as f
def dataframe_to_sql(df, table_name: str):
"""
将简单 PySpark DataFrame 操作（select + where）近似转换为 SQL 字符串。
⚠️ 注意：该函数基于逻辑计划字符串解析，非官方 API，仅适用于调试与开发辅助场景。
"""
plan = df._jdf.queryExecution().logical()
plan_str = plan.toString()
# 初始化 SQL 查询
sql = "SELECT "
# 提取 SELECT 列（匹配 Project [...] 结构）
if "Project" in plan_str:
start = plan_str.find("[", plan_str.find("Project")) + 1
end = plan_str.find("]", start)
if start > 0 and end > start:
select_part = plan_str[start:end].replace(" ", "").replace("\n", "")
# 简单清洗：移除别名（如 'a AS a' → 'a'）、函数包装（如 'unresolvedalias(a, None)'）
cleaned_cols = []
for col_expr in select_part.split(","):
col_expr = col_expr.strip()
# 移除 unresolvedalias(...) 包装
if col_expr.startswith("unresolvedalias("):
inner = col_expr[14:-1] # 去掉前缀和结尾括号
col_name = inner.split(",")[0].strip("'\"")
cleaned_cols.append(col_name)
else:
# 提取最内层标识符（如 'a', 'b', 'c' 或 'col(a)' → 'a'）
import re
match = re.search(r"[a-zA-Z_][a-zA-Z0-9_]*", col_expr)
if match:
cleaned_cols.append(match.group())
else:
cleaned_cols.append(col_expr)
sql += ", ".join(cleaned_cols)
else:
sql += "*"
else:
sql += "*"
# 添加 FROM 子句（需显式传入表名/路径，因逻辑计划中通常不保留原始路径）
sql += f" FROM `{table_name}`"
# 提取 WHERE 条件（匹配 Filter(...)）
if "Filter" in plan_str:
filter_start = plan_str.find("Filter") + len("Filter")
paren_start = plan_str.find("(", filter_start)
paren_end = -1
if paren_start != -1:
depth = 1
for i in range(paren_start + 1, len(plan_str)):
if plan_str[i] == '(':
depth += 1
elif plan_str[i] == ')':
depth -= 1
if depth == 0:
paren_end = i
break
if paren_start != -1 and paren_end != -1:
filter_expr = plan_str[paren_start + 1:paren_end].strip()
# 清洗表达式：还原列名、简化 like 比较等
filter_expr = filter_expr.replace("like", "LIKE").replace("unresolvedstar(*)", "*")
# 替换列引用：如 'a#123' → 'a'；'col(a)#124' → 'a'
import re
filter_expr = re.sub(r"([a-zA-Z_][a-zA-Z0-9_]*)#[0-9]+", r"\1", filter_expr)
filter_expr = re.sub(r"col\(([^)]+)\)", r"\1", filter_expr)
sql += f" WHERE {filter_expr}"
return sql
# 使用示例
PATH = "s3://my-bucket/my-table"
columns = ["a", "b", "c"]
data = spark.read.format("delta").load(PATH).select(*columns).where(f.col("a").like("%test%"))
sql = dataframe_to_sql(data, PATH)
print(sql)
# 输出示例：
# SELECT a, b, c FROM `s3://my-bucket/my-table` WHERE a LIKE %test%⚠️ 重要注意事项：
- 此方法依赖 df._jdf（Java DataFrame）及内部 queryExecution().logical()，属于 非公开、不稳定 API，可能随 Spark 版本升级而失效；
- 逻辑计划字符串格式无文档保证，不同 Spark 版本/优化器阶段输出差异较大，无法覆盖复杂操作（如 join、agg、window function、UDF 调用）；
- FROM 子句中的表名/路径必须由用户显式传入，因为逻辑计划中一般不保留原始数据源路径（尤其对 spark.read.load(...)）；
- 实际生产环境中，更推荐：
✅ 使用 df.explain(mode="extended") 查看逻辑/物理计划用于调试；
✅ 将业务逻辑统一用 SQL 编写（spark.sql("...")），再转为 DataFrame 处理；
✅ 对 Delta 表，直接使用 DESCRIBE DETAIL或 GENERATE SYNTAX 等元数据命令辅助开发。
综上，该方案适用于快速验证、教学演示或本地调试，切勿用于生产级 SQL 生成或自动化部署流程。
