PySpark中基于Spark SQL函数实现两列最长公共子串

我需要在PySpark中对两列执行最长公共子串（Longest Common Substring）对比。理论上可以用以下Python函数实现：

def longest_common_substring(input_string, string_list):
    longest_substrings = []
    for string in string_list:
        longest_substring = ""
        for i in range(len(input_string)):
            for j in range(i, len(input_string)):
                if string.startswith(input_string[i:j+1]):
                    if len(input_string[i:j+1]) > len(longest_substring):
                        longest_substring = input_string[i:j+1]
        longest_substrings.append(longest_substring)
    return longest_substrings

longest_common_substring("Node 1 - 2643", 
                         ['Node 1 - 2643', 'Node ', 'Node 1 - 2643 - Node 1 A032 - 32432'])

输出：

['Node 1 - 2643', 'Node ', 'Node 1 - 2643']

但实际数据量有1亿至数十亿行，性能至关重要，UDF不是最优方案。请问能否仅使用Spark SQL函数实现？

以下是示例数据及目标列lcs (goal)：

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("MWE - longest common substring") \
    .getOrCreate()

data = [
    ["AB1234|BC24412|DE34322", "AB1234|BC24412|DE9687", "AB1234|BC24412|DE"],
    ["AA2222|BB3333|CC6666", "AA2222|BD23123|EE12|ZZ929292", "AA2222|B"]
]
schema = ["pathA", "pathB", "lcs (goal)"]

df = spark.createDataFrame(data, schema=schema)

数据展示：

+----------------------+----------------------------+-----------------+
|pathA                 |pathB                       |lcs (goal)       | 
+----------------------+----------------------------+-----------------+
|AB1234|BC24412|DE34322|AB1234|BC24412|DE9687       |AB1234|BC24412|DE|
|AA2222|BB3333|CC6666  |AA2222|BD23123|EE12|ZZ929292|AA2222|B         |
+----------------------+----------------------------+-----------------+

针对大数据量场景，完全可以用Spark SQL内置函数实现，避免UDF的性能开销。以下是适配不同场景的方案：

方案1：前缀式公共子串（适配你的示例场景） #

你的示例中最长公共子串是从字符串开头的连续匹配部分，用递归CTE可以高效计算：

WITH recursive_lcs AS (
    SELECT 
        pathA,
        pathB,
        LEAST(LENGTH(pathA), LENGTH(pathB)) AS max_len,
        LEAST(LENGTH(pathA), LENGTH(pathB)) AS current_len
    FROM df
    UNION ALL
    SELECT 
        pathA,
        pathB,
        max_len,
        current_len - 1 AS current_len
    FROM recursive_lcs
    WHERE current_len > 0 
      AND SUBSTRING(pathA, 1, current_len) != SUBSTRING(pathB, 1, current_len)
)
SELECT 
    pathA,
    pathB,
    CASE 
        WHEN current_len = 0 THEN ''
        ELSE SUBSTRING(pathA, 1, current_len)
    END AS lcs
FROM recursive_lcs
WHERE current_len = 0 
   OR SUBSTRING(pathA, 1, current_len) = SUBSTRING(pathB, 1, current_len)
QUALIFY ROW_NUMBER() OVER (PARTITION BY pathA, pathB ORDER BY current_len DESC) = 1

说明 #

• 从两个字符串的最短长度开始，逐步缩短前缀长度，直到找到第一个匹配项
• 所有操作都是Spark原生向量化计算，性能远优于UDF
• 递归次数最多等于最短字符串的长度，资源消耗可控

方案2：通用最长公共子串（支持任意位置匹配） #

如果公共子串可能出现在字符串任意位置，可以用以下方案：

WITH string_lengths AS (
    SELECT 
        pathA,
        pathB,
        LENGTH(pathA) AS lenA,
        LENGTH(pathB) AS lenB
    FROM df
),
possible_lengths AS (
    SELECT 
        pathA,
        pathB,
        lenA,
        lenB,
        EXPLODE(SEQUENCE(LEAST(lenA, lenB), 1, -1)) AS candidate_len
    FROM string_lengths
),
substring_matches AS (
    SELECT 
        pathA,
        pathB,
        candidate_len,
        SUBSTRING(pathA, pos, candidate_len) AS subA,
        SUBSTRING(pathB, pos, candidate_len) AS subB
    FROM possible_lengths
    LATERAL VIEW SEQUENCE(1, lenA - candidate_len + 1) pos AS pos
    WHERE lenA >= candidate_len
    UNION ALL
    SELECT 
        pathA,
        pathB,
        candidate_len,
        SUBSTRING(pathA, pos, candidate_len) AS subA,
        SUBSTRING(pathB, pos, candidate_len) AS subB
    FROM possible_lengths
    LATERAL VIEW SEQUENCE(1, lenB - candidate_len + 1) pos AS pos
    WHERE lenB >= candidate_len
),
valid_matches AS (
    SELECT 
        pathA,
        pathB,
        candidate_len,
        subA AS lcs_candidate
    FROM substring_matches
    WHERE subA = subB
)
SELECT 
    pathA,
    pathB,
    COALESCE(FIRST(lcs_candidate), '') AS lcs
FROM valid_matches
QUALIFY ROW_NUMBER() OVER (PARTITION BY pathA, pathB ORDER BY candidate_len DESC) = 1

说明 #

• 先生成从最长到最短的所有可能子串长度
• 遍历所有对应长度的子串，筛选出两个字符串中相同的子串
• 最后取每个分组中最长的匹配结果
• 注意：长字符串会生成大量子串，建议在字符串长度可控时使用

方案3：针对|分隔路径的优化方案 #

你的数据是|分隔的路径片段，可先拆分片段再匹配，性能最优：

WITH split_paths AS (
    SELECT 
        pathA,
        pathB,
        SPLIT(pathA, '\\|') AS partsA,
        SPLIT(pathB, '\\|') AS partsB
    FROM df
),
full_part_matches AS (
    SELECT 
        pathA,
        pathB,
        CONCAT_WS('|', SLICE(partsA, 1, part_idx)) AS lcs_candidate,
        LENGTH(CONCAT_WS('|', SLICE(partsA, 1, part_idx))) AS match_len
    FROM split_paths
    LATERAL VIEW SEQUENCE(1, LEAST(SIZE(partsA), SIZE(partsB))) part_idx AS part_idx
    WHERE CONCAT_WS('|', SLICE(partsA, 1, part_idx)) = CONCAT_WS('|', SLICE(partsB, 1, part_idx))
),
partial_part_matches AS (
    SELECT 
        pathA,
        pathB,
        CONCAT(
            CONCAT_WS('|', SLICE(partsA, 1, LEAST(SIZE(partsA), SIZE(partsB)))),
            SUBSTRING(partsA[LEAST(SIZE(partsA), SIZE(partsB))], 1, pos)
        ) AS lcs_candidate,
        LENGTH(CONCAT(
            CONCAT_WS('|', SLICE(partsA, 1, LEAST(SIZE(partsA), SIZE(partsB)))),
            SUBSTRING(partsA[LEAST(SIZE(partsA), SIZE(partsB))], 1, pos)
        )) AS match_len
    FROM split_paths
    LATERAL VIEW SEQUENCE(1, LEAST(LENGTH(partsA[LEAST(SIZE(partsA), SIZE(partsB))]), LENGTH(partsB[LEAST(SIZE(partsA), SIZE(partsB))])) - 1) pos AS pos
    WHERE partsA[LEAST(SIZE(partsA), SIZE(partsB))] != partsB[LEAST(SIZE(partsA), SIZE(partsB))]
      AND SUBSTRING(partsA[LEAST(SIZE(partsA), SIZE(partsB))], 1, pos) = SUBSTRING(partsB[LEAST(SIZE(partsA), SIZE(partsB))], 1, pos)
),
all_matches AS (
    SELECT * FROM full_part_matches
    UNION ALL
    SELECT * FROM partial_part_matches
)
SELECT 
    pathA,
    pathB,
    COALESCE(FIRST(lcs_candidate), '') AS lcs
FROM all_matches
QUALIFY ROW_NUMBER() OVER (PARTITION BY pathA, pathB ORDER BY match_len DESC) = 1

说明 #

• 先按|拆分路径，匹配完整的前缀片段
• 再处理最后一个片段的部分匹配情况
• 大幅减少需要检查的子串数量，完全适配你的示例数据

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