从 Spark 中的数据框列值中删除空白 space

Question

我有一个模式的数据框（business_df）：

|-- business_id: string (nullable = true)
|-- categories: array (nullable = true)
|    |-- element: string (containsNull = true)
|-- city: string (nullable = true)
|-- full_address: string (nullable = true)
|-- hours: struct (nullable = true)
|-- name: string (nullable = true)

我想创建一个新数据框 (new_df)，以便 'name' 列中的值不包含任何空格。

我的代码是：

from pyspark import SparkContext
from pyspark.sql import SQLContext
from pyspark.sql import HiveContext
from pyspark.sql.functions import UserDefinedFunction
from pyspark.sql.types import StringType

udf = UserDefinedFunction(lambda x: x.replace(' ', ''), StringType())
new_df = business_df.select(*[udf(column).alias(name) if column == name else column for column in business_df.columns])
new_df.registerTempTable("vegas")
new_df.printSchema()
vegas_business = sqlContext.sql("SELECT stars, name from vegas limit 10").collect()

我一直收到这个错误：

NameError: global name 'replace' is not defined

这段代码有什么问题？

Answer 1

正如@zero323 所说，您可能在某处重叠了 replace 函数。我测试了您的代码，它运行良好。

from pyspark import SparkContext
from pyspark.sql import SQLContext
from pyspark.sql import HiveContext
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

df = sqlContext.createDataFrame([("aaa 111",), ("bbb 222",), ("ccc 333",)], ["names"])
spaceDeleteUDF = udf(lambda s: s.replace(" ", ""), StringType())
df.withColumn("names", spaceDeleteUDF("names")).show()

#+------+
#| names|
#+------+
#|aaa111|
#|bbb222|
#|ccc333|
#+------+

Answer 2

虽然您所描述的问题无法通过提供的代码重现，但使用 Python UDFs 来处理这样的简单任务效率相当低。如果您只想从文本中删除空格，请使用 regexp_replace:

from pyspark.sql.functions import regexp_replace, col

df = sc.parallelize([
    (1, "foo bar"), (2, "foobar "), (3, "   ")
]).toDF(["k", "v"])

df.select(regexp_replace(col("v"), " ", ""))

如果你想规范化空行使用trim:

from pyspark.sql.functions import trim

df.select(trim(col("v")))

如果你想保留前导/尾随空格，你可以调整regexp_replace:

df.select(regexp_replace(col("v"), "^\s+$", ""))

Answer 3

这是一个删除字符串中所有空格的函数：

import pyspark.sql.functions as F

def remove_all_whitespace(col):
    return F.regexp_replace(col, "\s+", "")

您可以这样使用函数：

actual_df = source_df.withColumn(
    "words_without_whitespace",
    quinn.remove_all_whitespace(col("words"))
)

remove_all_whitespace函数定义在quinn library中。 quinn 还定义了 single_space 和 anti_trim 方法来管理空格。 PySpark 定义了 ltrim、rtrim 和 trim 方法来管理空白。

Answer 4

我认为使用 regexp_replace 的解决方案即使对于少量数据也太慢了！所以我试图找到另一种方法，我想我找到了！

不漂亮，有点幼稚，但是速度很快！你怎么看？

def normalizeSpace(df,colName):

  # Left and right trim
  df = df.withColumn(colName,ltrim(df[colName]))
  df = df.withColumn(colName,rtrim(df[colName]))

  #This is faster than regexp_replace function!
  def normalize(row,colName):
      data = row.asDict()
      text = data[colName]
      spaceCount = 0;
      Words = []
      word = ''

      for char in text:
          if char != ' ':
              word += char
          elif word == '' and char == ' ':
              continue
          else:
              Words.append(word)
              word = ''

      if len(Words) > 0:
          data[colName] = ' '.join(Words)

      return Row(**data)

      df = df.rdd.map(lambda row:
                     normalize(row,colName)
                 ).toDF()
      return df
schema = StructType([StructField('name',StringType())])
rows = [Row(name='  dvd player samsung   hdmi hdmi 160W reais    de potencia 
bivolt   ')]
df = spark.createDataFrame(rows, schema)
df = normalizeSpace(df,'name')
df.show(df.count(),False)

打印

+---------------------------------------------------+
|name                                               |
+---------------------------------------------------+
|dvd player samsung hdmi hdmi 160W reais de potencia|
+---------------------------------------------------+

Answer 5

如@Powers 所示，有一个非常好用且易于阅读的函数来删除名为 quinn.You 的包提供的空格，可以在这里找到它：https://github.com/MrPowers/quinn Here are the instructions on how to install it if working on a Data Bricks workspace: https://docs.databricks.com/libraries.html

这里再次说明它是如何工作的：

#import library 
import quinn

#create an example dataframe
df = sc.parallelize([
    (1, "foo bar"), (2, "foobar "), (3, "   ")
]).toDF(["k", "v"])

#function call to remove whitespace. Note, withColumn will replace column v if it already exists
df = df.withColumn(
    "v",
    quinn.remove_all_whitespace(col("v"))
)

输出：

从 Spark 中的数据框列值中删除空白 space

Remove blank space from data frame column values in Spark

dataframe

apache-spark

apache-spark-sql

pyspark