数据帧到 RDD[行] 用空值替换空格

Question

我正在将Spark数据帧转换为RDD[Row]，以便我可以将其映射到最终模式以写入Hive Orc表。我想将输入中的任何空间转换为实际null，以便 hive 表可以存储实际null而不是空字符串。

输入数据帧（带有管道分隔值的单列）：

col1
1|2|3||5|6|7|||...|

我的代码：

inputDF.rdd.
  map { x: Row => x.get(0).asInstanceOf[String].split("\|", -1)}.
  map { x => Row (nullConverter(x(0)),nullConverter(x(1)),nullConverter(x(2)).... nullConverter(x(200)))}

def nullConverter(input: String): String = {
  if (input.trim.length > 0) input.trim
  else null
}

有没有干净的方法可以做到这一点，而不是调用 nullConverter 函数 200 次。

Answer 1

基于单列更新：

按照你的方法，我会做这样的事情：

inputDf.rdd.map((row: Row) => {
    val values = row.get(0).asInstanceOf[String].split("\|").map(nullConverter)
    Row(values)
})

---

将 nullConverter 或任何其他逻辑设为 udf：

import org.apache.spark.sql.functions._
val nullConverter = udf((input: String) => {
  if (input.trim.length > 0) input.trim
  else null
})

现在，在 df 上使用 udf 并应用于所有列：

val convertedDf = inputDf.select(inputDf.columns.map(c => nullConverter(col(c)).alias(c)):_*)

现在，您可以执行RDD逻辑。

Answer 2

在转换为RDD之前，使用DataFrame API会更容易做到这一点。首先，拆分数据：

val df = Seq(("1|2|3||5|6|7|8||")).toDF("col0")        // Example dataframe
val df2 = df.withColumn("col0", split($"col0", "\|")) // Split on "|"

然后找出数组的长度：

val numCols = df2.first.getAs[Seq[String]](0).length

现在，对于数组中的每个元素，使用 nullConverter UDF，然后将其分配给它自己的列。

val nullConverter = udf((input: String) => {
  if (input.trim.length > 0) input.trim
  else null
})
val df3 = df2.select((0 until numCols).map(i => nullConverter($"col0".getItem(i)).as("col" + i)): _*)

使用示例数据帧的结果：

+----+----+----+----+----+----+----+----+----+----+
|col0|col1|col2|col3|col4|col5|col6|col7|col8|col9|
+----+----+----+----+----+----+----+----+----+----+
|   1|   2|   3|null|   5|   6|   7|   8|null|null|
+----+----+----+----+----+----+----+----+----+----+

现在将其转换为RDD或继续将数据用作数据帧，具体取决于您的需求。

Answer 3

将数据帧转换为 rdd 是没有意义的

import org.apache.spark.sql.functions._
df = sc.parallelize([
    (1, "foo bar"), (2, "foobar "), (3, "   ")
]).toDF(["k", "v"])
df.select(regexp_replace(col("*"), " ", "NULL"))