Spark SQL中含嵌套括号的字符串递归拆分方案咨询

问题背景 #

需要对包含嵌套括号的Spark SQL字符串列进行递归式拆分提取，示例如下：

• 输入：A1 AND (A4 OR (A2 AND A3)) AND (B1 OR B2)
  预期输出：
  1. A2 AND A3->logic1
  2. A4 OR logic1 ->logic2
  3. B1 OR B2 ->logic3
  4. logic2 AND logic3 ->logic4
  5. A1 AND logic4 ->logic5
• 输入：(A2 AND A3) OR (B1 AND B2)
  预期输出：
  1. B1 AND B2->logic1
  2. A2 AND A3->logic2
  3. logic1 OR logic2->logic3

原方案通过split按(拆分结合dense_rank()排序，无法处理多层嵌套及复杂逻辑组合的场景，因为这种方式无法精准匹配成对的括号，也无法实现递归替换逻辑。

解决方案：自定义递归UDF实现括号匹配与替换 #

Spark SQL内置函数难以处理这种递归嵌套的字符串解析，推荐通过自定义UDF实现从最内层括号开始的递归提取与替换，同时记录每一步的拆分结果。

1. 编写递归处理逻辑（Scala版本） #

以下代码通过栈结构精准匹配最内层括号，递归完成表达式替换并收集每一步的拆分记录：

import org.apache.spark.sql.api.java.UDF1
import scala.collection.mutable.ListBuffer

class LogicSplitUDF extends UDF1[String, List[(String, String)]] {
  override def call(input: String): List[(String, String)] = {
    val steps = ListBuffer[(String, String)]()
    var currentStr = input
    var logicCount = 1

    def processInner(): Unit = {
      val stack = scala.collection.mutable.Stack[Int]()
      var startIdx = -1
      var endIdx = -1

      // 遍历字符串定位最内层括号对
      for (i <- currentStr.indices) {
        currentStr(i) match {
          case '(' =>
            stack.push(i)
            startIdx = i
          case ')' =>
            if (stack.nonEmpty) {
              stack.pop()
              // 栈空时找到最内层括号
              if (stack.isEmpty) {
                endIdx = i
                val innerExpr = currentStr.substring(startIdx + 1, endIdx).trim
                val logicTag = s"logic$logicCount"
                steps.append((innerExpr, logicTag))
                // 替换括号内容为logic标记
                currentStr = currentStr.substring(0, startIdx) + logicTag + currentStr.substring(endIdx + 1)
                logicCount += 1
                processInner()
                return
              }
            }
          case _ =>
        }
      }

      // 处理最后一层无括号的顶层逻辑
      if (currentStr.nonEmpty && steps.nonEmpty) {
        val lastLogicTag = s"logic$logicCount"
        steps.append((currentStr.trim, lastLogicTag))
      }
    }

    processInner()
    steps.toList
  }
}

// 注册UDF到SparkSession
spark.udf.register("split_logic", new LogicSplitUDF())

2. Spark SQL中调用UDF #

注册UDF后，通过explode展开拆分步骤，生成预期的分步结果：

WITH t1 AS (
    SELECT 'A1 AND (A4 OR (A2 AND A3)) AND (B1 OR B2)' AS logic
    UNION ALL
    SELECT '(A2 AND A3) OR (B1 AND B2)' AS logic
)
SELECT 
    original_logic,
    step,
    split_result
FROM (
    SELECT 
        logic AS original_logic,
        explode(split_logic(logic)) AS split_pair,
        row_number() OVER (PARTITION BY logic ORDER BY split_pair) AS step
    FROM t1
) t
LATERAL VIEW inline(array(split_pair)) AS expr, logic_tag
SELECT 
    original_logic,
    step,
    concat(expr, '->', logic_tag) AS split_result

3. 方案说明 #

• 括号匹配：用栈结构精准定位最内层括号，避免split函数导致的括号不匹配问题
• 递归替换：每完成一次最内层表达式替换，就记录原表达式->logicN的映射，再递归处理新生成的字符串
• 顶层处理：所有括号替换完成后，自动处理剩余的顶层逻辑组合，生成最后一步记录

注意事项 #

• 如果使用Python编写UDF，逻辑思路一致，但性能略低于Scala，适合数据量较小的场景
• 需确保输入的逻辑表达式括号是合法配对的，可在UDF中添加括号合法性检查逻辑

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