Spark SQL 中解析嵌套布尔表达式并递归提取逻辑子句的完整教程-Python教程-PHP中文网

Spark SQL 中解析嵌套布尔表达式并递归提取逻辑子句的完整教程

本文介绍如何在 Spark SQL 环境中准确解析含多层括号的布尔表达式（如 A1 AND (A4 OR (A2 AND A3))），并按运算优先级自底向上生成带编号的中间逻辑子句（如 A2 AND A3 → logic1），借助 Python 的 Lark 库实现健壮、可扩展的语法树解析与遍历。

本文介绍如何在 spark sql 环境中准确解析含多层括号的布尔表达式（如 `a1 and (a4 or (a2 and a3))`），并按运算优先级自底向上生成带编号的中间逻辑子句（如 `a2 and a3 → logic1`），借助 python 的 lark 库实现健壮、可扩展的语法树解析与遍历。

在 Spark SQL 原生函数（如 split()、regexp_replace() 或窗口函数）中直接处理深度嵌套、递归结构的布尔表达式存在根本性局限：它们缺乏语法感知能力，无法识别括号匹配、运算符优先级与表达式作用域。用户尝试基于括号数量排序分组的方式，在面对 (A1 AND A2) OR (A3 AND (A4 OR A5)) 等复杂结构时极易失效——因为括号层级不等于求值顺序，且 AND/OR 的结合性与嵌套位置共同决定执行次序。

真正可靠的解法是引入上下文无关文法（CFG）解析器，将字符串视为编程语言中的表达式进行结构化分析。Lark 是一个高性能、纯 Python 实现的解析库，支持 LALR(1) 和 Earley 解析算法，非常适合构建轻量级领域专用解析器（DSL parser）。

✅ 核心步骤概览

定义文法（Grammar）：用 BNF 风格描述布尔表达式的合法结构；
构建解析器与转换器（Transformer）：在语法树遍历过程中动态生成带编号的中间逻辑；
封装为 Spark UDF：将解析逻辑注册为 ArrayType(StringType()) 类型的 UDF，无缝集成进 DataFrame 流程；
展开与标准化输出：使用 explode() 展平数组，获得每条子逻辑的独立行记录。

? 文法设计说明

以下文法精准覆盖示例需求：

?start: expression
?expression: atom
    | expression "AND" expression -> and_op
    | expression "OR" expression  -> or_op
    | "(" expression ")" -> bracket_exp
?atom: /[A-Z][0-9]/ -> variable

%import common.WS
%ignore WS

?expression 表示左递归（支持 A AND B AND C 连续运算）；
-> and_op / -> or_op 为语义动作标记，便于 Transformer 区分节点类型；
bracket_exp 消除冗余括号，保留语义等价性；
variable 匹配形如 A1, B2 的原子标识符（可根据实际需要扩展为 [A-Za-z_][A-Za-z0-9_]*）。

⚙️ Transformer 实现关键逻辑

class MyTransformer(Transformer):
    def __init__(self):
        super().__init__()
        self.logic_counter = 0
        self.transformations = []  # 存储 (原始表达式, logicN) 元组

    def variable(self, items):
        return str(items[0])

    @v_args(inline=True)
    def and_op(self, left, right):
        self.logic_counter += 1
        result = f"{left} AND {right}"
        label = f"logic{self.logic_counter}"
        self.transformations.append((result, label))
        return label  # 返回 label 供父节点继续组合

    @v_args(inline=True)
    def or_op(self, left, right):
        self.logic_counter += 1
        result = f"{left} OR {right}"
        label = f"logic{self.logic_counter}"
        self.transformations.append((result, label))
        return label

    def bracket_exp(self, items):
        return items[0]  # 去括号，透传内部表达式结果

    def get_transformations(self):
        return [f"{orig} -> {lbl}" for orig, lbl in self.transformations]

? 注意：and_op 和 or_op 使用 @v_args(inline=True) 直接接收子节点值（而非列表），确保 left/right 已是已计算出的变量名或 logicN 标签，从而自然实现“自底向上”的求值序列。

? 完整 Spark 集成代码（可直接运行）

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, explode
from pyspark.sql.types import StructType, StructField, IntegerType, StringType, ArrayType
from pyspark.sql.functions import udf
from lark import Lark
from lark.visitors import Transformer
from lark.v_args import v_args

spark = SparkSession.builder.appName("BooleanExpressionParser").getOrCreate()

# 示例数据
schema = StructType([
    StructField("exp_id", IntegerType(), True),
    StructField("boolean_expression", StringType(), True),
])
df = spark.createDataFrame([
    (1, "A1 AND (A4 OR (A2 AND A3)) AND (B1 OR B2)"),
    (2, "(A2 AND A3) OR (B1 AND B2)"),
], schema)

# 文法与解析逻辑（同上）
grammar = """..."""  # 省略，见前文

def evaluate_exp(expression):
    class MyTransformer(Transformer):  # 同上实现
        ...
    parser = Lark(grammar, start='start', parser='lalr')
    tree = parser.parse(expression)
    transformer = MyTransformer()
    transformer.transform(tree)
    return transformer.get_transformations()

evaluate_exp_udf = udf(evaluate_exp, ArrayType(StringType()))
df_result = df.withColumn("ast_tree", evaluate_exp_udf(col("boolean_expression"))) \
              .withColumn("exploded_col", explode(col("ast_tree")))

df_result.select("exp_id", "exploded_col").show(truncate=False)

✅ 输出验证与注意事项

exp_id	exploded_col
1	A2 AND A3 -> logic1
1	A4 OR logic1 -> logic2
1	B1 OR B2 -> logic3
1	logic2 AND logic3 -> logic4
1	A1 AND logic4 -> logic5
2	A2 AND A3 -> logic1
2	B1 AND B2 -> logic2
2	logic1 OR logic2 -> logic3