PySpark实现关键词匹配替换：迁移Pandas逻辑至Spark环境

Got it, let's walk through how to migrate this keyword replacement logic from Pandas to PySpark. Here's a practical, efficient approach tailored to your use case:

Step 1: Set up the keyword-to-lookupid mapping #

First, we'll turn your Python lists into a PySpark DataFrame—this makes it easy to generate our replacement rules. We'll also handle case insensitivity since your example title uses lowercase "store manager" but the keyword is "Store Manager".

import re
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from functools import reduce

# Initialize Spark session (skip if you already have one running)
spark = SparkSession.builder.appName("KeywordReplacement").getOrCreate()

# Your provided mapping data
keywords = ['IT Manager', 'Sales Manager', 'IT Analyst', 'Store Manager']
lookupid = ['##10##','##13##','##12##','##13##']

# Create a structured mapping DataFrame
mapping_df = spark.createDataFrame(zip(keywords, lookupid), schema=["keyword", "lookupid"])

Step 2: Read your title CSV file #

Next, load the CSV containing the title column you need to process:

# Replace with your actual file path
title_df = spark.read.csv("path/to/your/titles.csv", header=True, inferSchema=True)

Step 3: Apply keyword replacements efficiently #

We'll use a chain of regexp_replace functions to substitute each keyword with its corresponding lookupid. The (?i) flag ensures the replacement is case-insensitive, and re.escape() handles any special characters in your keywords (like spaces) so they don't break the regex pattern.

# Generate a list of replacement expressions for each keyword-lookupid pair
replacements = [
    F.regexp_replace("title", rf"(?i){re.escape(row.keyword)}", row.lookupid)
    for row in mapping_df.collect()
]

# Chain all replacements together using reduce to update the title column iteratively
final_df = reduce(lambda df, expr: df.withColumn("title", expr), replacements, title_df)

# Verify the results (truncate=False shows full text)
final_df.show(truncate=False)

Key Notes: #

• Case Insensitivity: The (?i) in the regex ensures matches work regardless of uppercase/lowercase in the title (e.g., "Store Manager", "store manager", "STORE MANAGER" all get replaced correctly).
• Special Character Safety: re.escape() escapes any characters in keywords that have special meaning in regex (like spaces, parentheses, etc.), so your replacements don't fail unexpectedly.
• Performance: For small keyword lists like your example, using collect() to build the replacement chain is totally fine. If you're working with a very large mapping dataset, a broadcast join approach would be more efficient—just let me know if you need that adjustment!

Testing with your example title:

"I have been working here as a store manager since after I passed f..."

It would get converted to:

"I have been working here as a ##13## since after I passed f..."

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