你可以用这样的东西来实现它，

# First calculate the number of "2"s from the start till some other value is found, for eg '2223' should give you 3 as the length
# Use that calculated value to repeat the "9" that many times
# replace starting "2"s with the calulated "9" string
# finally drop all the calculated columns
df.withColumn("len_2", F.length(F.regexp_extract(F.col("value"), r"^2*(?!2)", 0)).cast('int'))\
    .withColumn("to_replace_with", F.expr("repeat('9', len_2)"))\
    .withColumn("new_value", F.expr("regexp_replace(value, '^2*(?!2)', to_replace_with)")) \
    .drop("len_2", "to_replace_with")\
    .show(truncate=False)

输出：

+-------------+-------------+
|value        |new_value    |
+-------------+-------------+
|2222238308750|9999938308750|
|222222579844 |999999579844 |
|222225701296 |999995701296 |
|2222250087899|9999950087899|
|2222250087899|9999950087899|
|2222237274658|9999937274658|
|22222955099  |99999955099  |
|22222955099  |99999955099  |
|22222955099  |99999955099  |
|222285678    |999985678    |
+-------------+-------------+

我使用的列名为value，您必须将其替换为id。

您可以尝试以下操作：

from pyspark.sql.functions import *

df = df.withColumn("tempcol1", regexp_extract("id", "^2*", 0)).withColumn("tempcol2", split(regexp_replace("id", "^2*", "_"), "_")[1]).withColumn("new_id", concat((regexp_replace("tempcol1", "2", "9")), "tempcol2")).drop("tempcol1", "tempcol2")

id列被拆分为两个临时列，一个具有前缀，另一个具有字符串的其余部分。前缀列的值被替换，并与第二个临时列连接回去。