在火花数据框中用空白值替换空值不起作用答案

【问题标题】：Replacing null value with blank value in spark data frame not working在火花数据框中用空白值替换空值不起作用
【发布时间】：2017-10-09 06:23:54
【问题描述】：

我有两个数据集，数据集 1 在下面

LineItem.organizationId|^|LineItem.lineItemId|^|StatementTypeCode|^|LineItemName|^|LocalLanguageLabel|^|FinancialConceptLocal|^|FinancialConceptGlobal|^|IsDimensional|^|InstrumentId|^|LineItemSequence|^|PhysicalMeasureId|^|FinancialConceptCodeGlobalSecondary|^|IsRangeAllowed|^|IsSegmentedByOrigin|^|SegmentGroupDescription|^|SegmentChildDescription|^|SegmentChildLocalLanguageLabel|^|LocalLanguageLabel.languageId|^|LineItemName.languageId|^|SegmentChildDescription.languageId|^|SegmentChildLocalLanguageLabel.languageId|^|SegmentGroupDescription.languageId|^|SegmentMultipleFundbDescription|^|SegmentMultipleFundbDescription.languageId|^|IsCredit|^|FinancialConceptLocalId|^|FinancialConceptGlobalId|^|FinancialConceptCodeGlobalSecondaryId|^|FFAction|!|
Japan|^|1507101869432|^|4295876606|^|1|^|BAL|^|Cash And Deposits|^|null|^|null|^|ACAE|^|false|^|null|^|null|^|null|^|null|^|false|^|null|^|null|^|null|^|null|^|505126|^|505074|^|null|^|null|^|null|^|null|^|null|^|null|^|null|^|3018759|^|null|^|I|!|

这就是我使用自动发现模式加载数据的方式

val df1With_ = df.toDF(df.columns.map(_.replace(".", "_")): _*)
val column_to_keep = df1With_.columns.filter(v => (!v.contains("^") && !v.contains("!") && !v.contains("_c"))).toSeq
val df1result = df1With_.select(column_to_keep.head, column_to_keep.tail: _*)

数据集 2：

4295867927|^|860|^|CUS|^|External Revenue|^||^||^|REXR|^|False|^||^||^||^||^|False|^|False|^|CUS_REXR|^||^||^|505074|^|505074|^|505074|^|505074|^|505074|^||^|505074|^|True|^||^|3015250|^||^|I|!|

我从两者中创建一个数据框，然后加入。最后，我将两个数据帧的输出写入 csv 文件。

这是写入 csv 文件的代码。

val dfMainOutputFinal = dfMainOutput.select($"DataPartition", $"StatementTypeCode",concat_ws("|^|", dfMainOutput.schema.fieldNames.filter(_ != "DataPartition").map(c => col(c)): _*).as("concatenated"))

val dfMainOutputFinalWithoutNull = dfMainOutputFinal.withColumn("concatenated", regexp_replace(dfMainOutputFinal.col_*, "null", "")).show()

dfMainOutputFinal.write.partitionBy("DataPartition","StatementTypeCode")
  .format("csv")
  .option("nullValue", "")
  .option("codec", "gzip")
  .save("s3://trfsdisu/SPARK/FinancialLineItem/output")

除.option("nullValue", "") 外，其他一切工作正常。我无法将 null 替换为空白值。

在我的输出中我仍然看到空值。

我也试过了，结果一样。

val newDf = df.na.fill("e",Seq("blank"))

【问题讨论】：

您确定数据集包含实际的 nulls 而不是字符串“null”吗？
所以它会只替换 null 而不是“null”吗？如果我有空字符串怎么办...
null 的字符串很好，但如果您有一个带有字母“null”的字符串，它会被视为普通字符串。
如果您实际上有一个字符串“null”，一个选项是简单地将所有出现的“null”替换为“”。
所以在理智的选项中代替 nullValue 我可以放 null 吗？或者我必须创建另一个数据框，该数据框源自第一个替换 null

标签： scala apache-spark spark-dataframe spark-csv

【解决方案1】：

我怀疑数据框实际上并不包含 nulls，而是它们是带有字母“null”的字符串。如果是这种情况，那么您可以简单地将所有“null”实例替换为“”。在此之后，您可以像以前一样使用.option("nullValue", "")。要替换列中的字符串，可以使用regexp_replace(column, "string to replace", "string to replace with")。 S小例子：

val df = Seq("a", "null", "c", "b").toDF("col1")
val df2 = df.withColumn("col1", regexp_replace(col("col1"), "null", ""))

这里的“null”被替换为“”，最终的数据框如下所示：

+----+
|col1|
+----+
|   a|
|    |
|   c|
|   b|
+----+

【讨论】：

Si 在我的情况下，我必须对所有列都这样做吗？我可以在写入 csv 文件之前这样做吗？
@SUDARSHAN 是的，您需要在写入 csv 文件之前对所有列执行此操作。
那么架构会再次发生变化，我在写入分区文件时会遇到问题
@SUDARSHAN 您不必对架构进行任何更改。正如我上面的“col1”列示例，输入和输出列是相同的。因此，最终的数据框具有相同的列名和相同的列类型。
@SUDARSHAN 查看以前的问题，我假设您在编写 csv 之前自己合并列。最简单的方法是仅在最后的合并列上执行此步骤（使用regexp_replace）。然后你只需要做一次。

【解决方案2】：

option("nullValue", "whatever") 检查是否有任何列值“随便”，并将该列值视为数据框中的空值。

在阅读过程中使用该选项就可以了。

 Dataset<Row> df = spark.read().format("csv")
              .option("nullValue", "NULL")      // this config does the trick
              .option("sep", ",")
              .schema(structType)
              .load(filePath);

【讨论】：