如何在PySpark中从指定格式文件名提取目标子串并写入表列？

问题背景 #

我们通过以下命令从ADLS读取文件：

relative_path = "ContainerName/"
input_file_name = "ss_old_data_Amazon_Blast_202211093837474.csv"

需要从待处理的文件名中提取特定子串，具体要求如下：

• 移除前缀（多数文件以vm_path开头，部分以ss_old_data开头，忽略大小写）
• 排除日期数字部分和.csv后缀
• 移除子串中的下划线，同时去掉括号及内部内容，最后转为大写，仅保留主体相关内容

示例参考 #

PySpark实现代码 #

通过组合PySpark内置字符串函数即可完成需求，具体代码如下：

from pyspark.sql import SparkSession
from pyspark.sql.functions import regexp_replace, upper, col

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

# 构造示例数据
sample_data = [
    ("vm_Path_Accenture_Complex_Union_202211027373.csv",),
    ("vm_path_Google_is_a_good_company_20221109473.csv",),
    ("ss_old_data_Amazon_Blast_202211093837474.csv",),
    ("ss_old_data_Black_Adam_(TY)_2022847093837474.csv",),
    ("ss_old_data_Man_of_steel_(PQ)_2022847093837474.csv",)
]

df = spark.createDataFrame(sample_data, ["file_name"])

# 核心处理逻辑
result_df = df.withColumn(
    "extracted_content",
    upper(
        regexp_replace(
            regexp_replace(
                regexp_replace(
                    regexp_replace(col("file_name"), r"\.csv$", ""),  # 移除.csv后缀
                    r"^(vm_path|ss_old_data)_", "", 0, "i"  # 移除开头前缀，忽略大小写
                ),
                r"_\d+$", ""  # 移除末尾的下划线+日期数字部分
            ),
            r"_\(.*?\)|_", ""  # 移除下划线，以及下划线+括号内的内容
        )
    )
)

# 输出结果
result_df.show(truncate=False)

代码逻辑说明 #

1. regexp_replace(col("file_name"), r"\.csv$", "")：精准匹配文件名末尾的.csv并移除
2. regexp_replace(..., r"^(vm_path|ss_old_data)_", "", 0, "i")：匹配开头的vm_path_或ss_old_data_，"i"参数实现大小写不敏感匹配，直接移除该前缀
3. regexp_replace(..., r"_\d+$", "")：匹配文件名末尾的下划线+连续数字序列，移除这部分日期内容
4. regexp_replace(..., r"_\(.*?\)|_", "")：同时匹配两种模式——下划线+括号及内部内容（如_(TY)）、单独的下划线，全部移除
5. upper(...)：将处理后的字符串统一转为大写

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