使用 regexp_replace 在 pyspark 上循环出现错误消息答案

【问题标题】：Error message in a loop for on pypsark using regexp_replace使用 regexp_replace 在 pyspark 上循环出现错误消息
【发布时间】：2020-07-09 14:58:33
【问题描述】：

我正在 pyspark 中创建一个循环，并且我收到以下消息：

"Column is not iterable"

这是代码：

(regexp_replace(data_join_result[varibale_choisie],
(random.choice(data_join_result.collect()[j][varibale_choisie])),
data_join_result.collect()[j][lettre_choisie] ))))

在错误信息中，问题就在此时出现：

data_join_result.collect()[j][lettre_choisie]

如果有人知道如何解决它！谢谢

【问题讨论】：

你能添加一个示例输入和预期输出吗
刚刚添加，谢谢！
所以你只需要在你的列中随机替换？
是的，一个随机字母被随机列中的另一个随机字母替换（我试图破坏我的数据集）
您需要替换一个字符吗？还是可以完全打乱这个词？

标签： pyspark extract-error-message

【解决方案1】：

不建议在驱动程序中收集数据，还要遍历数据帧。 Spark 提供了多个 api，允许我们以并行方式执行我们的任务。在您的情况下，您可以尝试以下方法：

对于单个字符替换，试试这个（性能密集型）选项

import pyspark.sql.functions as F
import string
import random
test1 = spark.createDataFrame([("Mike","apple", "oranges", "red wine"),("Kate","Whitewine", "green beans", "waterrr"), ("Leah", "red wine","juice","rice")],schema=["col1","col2","col3","col4"])
cols = test1.columns
alp=(list(string.ascii_lowercase))  

@F.udf(test1.schema)
    def randomize(row):
        row_d = row.asDict()
        pos_sel = random.randint(0,len(cols)-1)    
        col_select = cols[pos_sel]
        row_d[col_select]=row_d[col_select].replace(alp[random.randint(1,24)],alp[random.randint(1,24)],1)
        return(row_d)
    
    test2 = test1.withColumn("struct_coln",randomize(F.struct(cols))).select('struct_coln.*')

结果：

+----+---------+-----------+--------+
|col1|col2     |col3       |col4    |
+----+---------+-----------+--------+
|Mike|apple    |orangos    |red wine|
|Kate|Whitewine|green beans|waterrr |
|Leah|red wine |juice      |rice    |
+----+---------+-----------+--------+

您可以看到橙子被损坏为橙子。如果您将要替换的字母限制为仅元音，则会增加损坏的可能性。

如果您不需要替换一个字符，请尝试以下操作：

test1 = spark.createDataFrame([("Mike","apple", "oranges", "red wine"),("Kate","Whitewine", "green beans", "waterrr"), ("Leah", "red wine","juice","rice")],schema=["col1","col2","col3","col4"])
cols = test1.columns
alp=(list(string.ascii_lowercase))


#%%"
for i in range(30):    
    pos_sel = random.randint(0,len(cols)-1)    
    col_select = cols[pos_sel]
    tst_rep = test1.withColumn(col_select,F.translate(F.col(col_select),alp[random.randint(1,24)],alp[random.randint(1,24)]))
    test1 = tst_rep

在这里，您可以通过控制循环迭代来进行一些控制

结果：

test1.show()
+----+---------+-----------+--------+
|col1|     col2|       col3|    col4|
+----+---------+-----------+--------+
|Mike|    applu|    oranges|rjd winj|
|Kate|Whifuwinu|green beans| watjrrr|
|Leah| rud winu|      juihe|    ricj|
+----+---------+-----------+--------+

【讨论】：

你好，谢谢，这对我有帮助！实际上，我还有一个问题：ValueError: D 。你知道我该如何解决吗？（我需要，我可以发布整个代码）
不客气。当你尝试上面的代码时你得到了错误？还是和别的东西有关？
用别的东西，我正在制作一个循环来破坏数据框，所以“替换”是我想要创建的错误之一
为了保持问题和答案的简洁，您能否在此处关闭此问题（最好接受并投票 - 没有义务）并将您的新问题作为新问题发布？带有示例输入数据、预期输出和您尝试过的内容。
我做到了，我的问题是：“PySpark 中数据框的损坏循环”。（你看到后我会删除这条评论）。Thx

【解决方案2】：

>最后，我找到了如何创建一个**循环来破坏数据集**。如果有人需要一天，我会分享！

首先，您需要定义要创建的错误，用于替换的字母，例如要损坏的变量，然后添加带有特殊字符的错误：

lettre = [ "A", "B", "C", "D", "E", "F", "G", "H", "I", "J", "K", "L", "M", "N", "O", "P", "Q", "R", "S", "T", "U", "V", "W", "X", "Y", "Z"]

code_erreur= [ "replace","inserte","delete","espace","caract_spe", "NA","inverse"]

nombre_erreur=["1","1","1","2"]

varibale =["VARIABLEA","VARIABLEB"]

caract_spe =["_", "^", "¨", "", ".", "é", "-", "*","ù","ï","à","è","î","â"]

我创建了一个列表“nombre_erreur”，因为我想要 75% 的数据集有 1 个错误，而 25% 的数据集有 2 个错误。

接下来，创建定义：

def def_code_erreur(code_erreur,varibale ,nombre_erreur,lettre,caract_spe):

  if type_erreur=="delete":
    for i in range(0,int(nb_erreur)):
      longueur = len(col1)
      pos = random.choice(range(1,longueur))
      col1 = col1[:pos] + col1[(pos+1):]
      
  if type_erreur=="espace":
    for i in range(0,int(nb_erreur)):
      longueur = len(col1)
      pos = random.choice(range(1,longueur))
      col1 = col1[:pos] + " " + col1[(pos):]
      
  if type_erreur=="inserte":
    for i in range(0,int(nb_erreur)):
      longueur = len(col1)
      pos = random.choice(range(1,longueur))
      col1 = col1[:pos] + lettre_choisie + col1[(pos):] 
      
  if type_erreur=="caract_spe":
    for i in range(0,int(nb_erreur)):
      longueur = len(col1)
      pos = random.choice(range(1,longueur))
      col1 = col1[:pos] + caract_spe_choisi + col1[(pos):]
      
  if type_erreur=="replace":
    for i in range(0,int(nb_erreur)):
      longueur = len(col1)
      pos = random.choice(range(1,longueur))
      col1 = col1[:pos-1] + lettre_choisie + col1[(pos):]      
      
  if type_erreur=="inverse":
    for i in range(0,int(nb_erreur)):
      longueur = len(col1)
      pos = random.choice(range(1,longueur))
      col1 = col1[:pos-1] + col1[pos:pos+1] + col1[pos-1:pos] + col1[(pos+1):]      
      
  if type_erreur=="NA":
    for i in range(0,int(nb_erreur)):
      col1 = col1

    
  return col1


udf_def_code_erreur = udf(def_code_erreur, StringType())

没关系，你必须调用“udf_def_code_erreur”！！如果要破坏整个数据集，可以循环调用它。

【讨论】：