Scala/PySpark如何按模式解析格式化字符串（类SQL Server实现）

问题分析 #

你遇到的StringIndexOutOfBoundsException是因为手动使用substring时，硬编码的索引超过了原字符串的实际长度。要实现动态模式匹配的字符串格式化，核心是先解析模式中的分段长度，再按规则截取拼接，而非依赖固定索引。

Scala 实现方案 #

核心思路

1. 解析模式：提取模式中每个X段的长度，同时保留分隔符（如-）
2. 按分段长度依次截取原字符串的对应部分
3. 用分隔符拼接各段，得到格式化结果

代码示例

object StringFormatter {
  // 解析模式，返回(分段长度列表, 分隔符)
  private def parsePattern(pattern: String): (List[Int], String) = {
    val segLengths = pattern.split("[^X]").filter(_.nonEmpty).map(_.length).toList
    val separator = pattern.replaceAll("X", "").distinct // 假设模式使用统一分隔符
    (segLengths, separator)
  }

  // 格式化字符串
  def formatString(input: String, pattern: String): String = {
    val (segLengths, separator) = parsePattern(pattern)
    var currentPos = 0
    val segments = segLengths.map { len =>
      val endPos = math.min(currentPos + len, input.length)
      val seg = input.substring(currentPos, endPos)
      currentPos = endPos
      seg
    }
    segments.mkString(separator)
  }
}

// 测试
val input = "abcdefg"
println(StringFormatter.formatString(input, "XX-XXX-XX")) // 输出 ab-cde-fg
println(StringFormatter.formatString(input, "X-XX-XXXX")) // 输出 a-bc-defg

PySpark 实现方案 #

PySpark中可通过自定义UDF封装上述逻辑，直接应用到DataFrame列上。

代码示例

from pyspark.sql import SparkSession
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

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

def parse_pattern(pattern):
    seg_lengths = [len(part) for part in pattern.split("[^X]") if part]
    separator = pattern.replace("X", "").strip()
    return seg_lengths, separator

def format_string(input_str, pattern):
    seg_lengths, separator = parse_pattern(pattern)
    current_pos = 0
    segments = []
    for len_ in seg_lengths:
        end_pos = min(current_pos + len_, len(input_str))
        segments.append(input_str[current_pos:end_pos])
        current_pos = end_pos
    return separator.join(segments)

# 注册UDF
format_string_udf = udf(format_string, StringType())

# 测试DataFrame
data = [("abcdefg", "XX-XXX-XX"), ("abcdefg", "X-XX-XXXX")]
df = spark.createDataFrame(data, ["input_str", "pattern"])
df.withColumn("formatted_str", format_string_udf(df["input_str"], df["pattern"])).show()

# 输出结果
# +---------+-----------+-------------+
# |input_str|    pattern|formatted_str|
# +---------+-----------+-------------+
# | abcdefg|XX-XXX-XX|    ab-cde-fg|
# | abcdefg|X-XX-XXXX|     a-bc-defg|
# +---------+-----------+-------------+

错误原因说明 #

你之前的代码出现越界异常，是因为直接固定了substring的结束索引（比如写到10），但原字符串长度只有7。通过动态计算每个分段的起始和结束位置，确保不超过字符串实际长度，就能彻底避免这类问题。

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