使用函数返回新的Dataframe（通过转换现有的Dataframe） - spark / scala

Question

我是Spark的新手。我试图将JSONArray读入Dataframe并对其进行一些转换。我试图通过删除一些html标签和一些newline字符来清理我的数据。例如：

从JSON读取的初始数据帧：

+-----+---+-----+-------------------------------+
|index|  X|label|      date                     |
+-----+---+-----+-------------------------------+
|    1|  1|    A|<div>&quot2017-01-01&quot</div>|
|    2|  3|    B|<div>2017-01-02</div>          |
|    3|  5|    A|<div>2017-01-03</div>          |
|    4|  7|    B|<div>2017-01-04</div>          |
+-----+---+-----+-------------------------------+

应该转变为：

+-----+---+-----+------------+
|index|  X|label|      date  |
+-----+---+-----+------------+
|    1|  1|    A|'2017-01-01'|
|    2|  3|    B|2017-01-02  |
|    3|  5|    A|2017-01-03  |
|    4|  7|    B|2017-01-04  |
+-----+---+-----+------------+

我知道我们可以使用以下方法执行这些转换：

df.withColumn("col_name",regexp_replace("col_name",pattern,replacement))

我可以使用withColumn清理我的数据，如上所示。但是，我有大量的列，并为每列编写.withColumn方法似乎不优雅，简洁或高效。所以我尝试做这样的事情：

  val finalDF = htmlCleanse(intialDF, columnsArray)

  def htmlCleanse(df: DataFrame, columns: Array[String]): DataFrame = {
    var retDF = hiveContext.emptyDataFrame
    for(i <- 0 to columns.size-1){
      val name = columns(i)
      retDF = df.withColumn(name,regexp_replace(col(name),"<(?:\"[^\"]*\"['\"]*|'[^']*'['\"]*|[^'\">])+>",""))
                .withColumn(name,regexp_replace(col(name),"&quot;","'"))
                .withColumn(name,regexp_replace(col(name),"&#160;"," "))
                .withColumn(name,regexp_replace(col(name),"&#58;",":"))
    }
    retDF
  }

我定义了一个新函数htmlCleanse，我将要转换的Dataframe和列数组传递给函数。该函数创建一个新的emptyDataFrame并迭代列列表执行单个迭代的列清理并将转换后的df分配给retDF变量。

这给了我没有错误，但它似乎没有从所有列中删除html标签，而一些列似乎被清理。不确定这种不一致行为的原因是什么（对此有任何想法？）。

那么，什么是清理我的数据的有效方法？任何帮助，将不胜感激。谢谢！

Answer 1

第一个问题是初始化空框架什么都不做，你只需创建新的东西。然后，你不能在没有连接的情况下从另一个数据框“添加”它（这在性能方面是个坏主意）。

第二个问题是retDF总是从df定义。这意味着你扔掉了你所做的一切，除了清理最后一列。

相反，您应该将retDF初始化为df，并在每次迭代中修复一个列并覆盖retDF，如下所示：

def htmlCleanse(df: DataFrame, columns: Array[String]): DataFrame = {
    var retDF = df 
    for(i <- 0 to columns.size-1){
      val name = columns(i)
      retDF = retDF.withColumn(name,regexp_replace(col(name),"<(?:\"[^\"]*\"['\"]*|'[^']*'['\"]*|[^'\">])+>",""))
                   .withColumn(name,regexp_replace(col(name),"&quot;","'"))
                   .withColumn(name,regexp_replace(col(name),"&#160;"," "))
                   .withColumn(name,regexp_replace(col(name),"&#58;",":"))
    }
    retDF
  }