空值

Question

我试图插入数据帧到卡桑德拉：

result.rdd.saveToCassandra(keyspaceName, tableName) 

然而，一些列的值是空的，因此我得到异常：

java.lang.NumberFormatException: empty String 
at sun.misc.FloatingDecimal.readJavaFormatString(FloatingDecimal.java:1842) 
at sun.misc.FloatingDecimal.parseFloat(FloatingDecimal.java:122) 
at java.lang.Float.parseFloat(Float.java:451) 
at scala.collection.immutable.StringLike$class.toFloat(StringLike.scala:231) 
at scala.collection.immutable.StringOps.toFloat(StringOps.scala:31) 
at com.datastax.spark.connector.types.TypeConverter$FloatConverter$$anonfun$convertPF$4.applyOrElse(TypeConverter.scala:216) 

有没有一种办法用数据框中的空值替换所有的EMPTY值，是否可以解决这个问题？ 对于这个问题，让我们假设这是数据帧DF：

col1 | col2 | col3 
"A" | "B" | 1 
"E" | "F" | 
"S" | "K" | 5 

我如何可以替换与空COL3空值？

Answer 1

如果您将DataFrame列转换为数字类型，则任何无法归为相应类型的值都将变为空值。

import org.apache.spark.sql.types.IntegerType 
df.select(
    $"col1", 
    $"col2", 
    $"col3" cast IntegerType 
) 

，或者如果你没有一个select语句

df.withColumn("col3", df("col3") cast IntegerType) 

如果要将此应用到，觉得它会做太不方便做这在SELECT语句中的列或者如果投射不适合您的情况，您可以转换为rdd以应用转换，然后返回到数据框。你可能想为此定义一个方法。

def emptyToNull(df: DataFrame): DataFrame = { 
    val sqlCtx = df.sqlContext 
    val schema = df.schema 

    val rdd = df.rdd.map(
     row => 
     row.toSeq.map { 
      case "" => null 
      case otherwise => otherwise 
     }) 
     .map(Row.fromSeq) 

    sqlCtx.createDataFrame(rdd, schema) 
    } 

Answer 2

你可以写这样的udf：

val df = Seq(("A", "B", "1"), ("E", "F", ""), ("S", "K", "1")).toDF("col1", "col2", "col3") 
// make a udf that converts String to option[String] 
val nullif = udf((s: String) => if(s == "") None else Some(s)) 

df.withColumn("col3", nullif($"col3")).show 

+----+----+----+ 
|col1|col2|col3| 
+----+----+----+ 
| A| B| 1| 
| E| F|null| 
| S| K| 1| 
+----+----+----+ 

---

您还可以使用when.otherwise，如果你想避免的UDF用法：

df.withColumn("col3", when($"col3" === "", null).otherwise($"col3")).show 

+----+----+----+ 
|col1|col2|col3| 
+----+----+----+ 
| A| B| 1| 
| E| F|null| 
| S| K| 1| 
+----+----+----+ 

---

或者您可以使用SQL nullif函数进行转换ert空字符串为空：

df.selectExpr("col1", "col2", "nullif(col3, \"\") as col3").show 
+----+----+----+ 
|col1|col2|col3| 
+----+----+----+ 
| A| B| 1| 
| E| F|null| 
| S| K| 1| 
+----+----+----+