Как исправить `` Имя атрибута Pyspark содержит недопустимый символ`

Question

У меня есть pandas df, который я конвертирую в spark для сохранения.

При запуске

clean_bows_df.coalesce(1).write.mode("overwrite").parquet(params["output_clustered_data_path"])

я сталкиваюсь со следующим:

org.apache.spark.sql.AnalysisException: Attribute name "onehot_tag_user_me al" contains invalid character(s) among " ,;{}()\n\t=". Please use alias to rename it.;

Я создаю spark df, используя:

  clean_bows_df = spark.createDataFrame(df_pd)      
#  clean_bows_df.select("*", f.regexp_replace(f.col("*"), "[&#$#,]", ""))
  clean_bows_df.replace('&#',"")
#   clean_bows_df.regexp_replace('[\&#]',"")

Комментированные строки были попытками заменить оскорбительные символы. Имена столбцов генерируются из данных df, поэтому я заранее не знаю, какими они будут. Еще в pandas я запускаю

  colnames = list(df_pd.columns.values)  
  colnames = [col.replace('&#34','_') for col in colnames] 
  colnames = [col.replace('&quot','_') for col in colnames] 
  colnames = [col.replace('&','_') for col in colnames] 
  colnames = [col.replace('quot','_') for col in colnames] 

и

def replace_substrings(s):
  s = re.sub(r'&#', ' ', s)  # attempt to fix spark conversion error 'cannot resolve &#39 given columns...'
  s = re.sub(r'#', ' ', s)  # attempt to fix spark conversion error 'cannot resolve &#39 given columns...'
  s = re.sub(r'&', ' ', s)  # attempt to fix spark conversion error 'cannot resolve &#39 given columns...'
  s = re.sub(r'[ ,;{}()\\n\\t=\n\t]+', ' ',s )
  s = s.replace("&#34","");      
  s = s.replace('(',' ')   #  ,;{}()\n\t=". 
  s = s.replace(')',' ')   #  ,;{}()\n\t=". 
  s = s.replace('{',' ')   #  ,;{}()\n\t=". 
  s = s.replace('}',' ')   #  ,;{}()\n\t=". 
  s = s.replace('\n',' ')   #  ,;{}()\n\t=". 
  s = s.replace('\t',' ')   #  ,;{}()\n\t=". 
  s = s.replace('=',' ')   #  ,;{}()\n\t=". 
  s = s.replace(',',' ')   #  ,;{}()\n\t=". 
  s = s.replace(';',' ')   #  ,;{}()\n\t=". 
  s = s.replace("&quot","");    
  return s

df_pd = df_pd.applymap(lambda s:replace_substrings(s) if type(s) == str else s)

Я не слишком знаком с искрой, поэтому, если кто-то может подсказать мне, как заменить все оскорбительные последовательности в все имена столбцов я был бы обязан.