空数据类型 - mongo Spark 阅读器

【问题标题】：null datatype - mongo Spark Reader空数据类型 - mongo Spark 阅读器
【发布时间】：2021-07-27 13:07:44
【问题描述】：

全部，

我正在使用 spark 读取 mongo 连接，其中一些数据类型为 null；我在pyspark中找到了用字符串替换空数据类型的代码，但在scala中需要类似- 谁能帮忙怎么做？

def fix_spark_schema(schema):
  if schema.__class__ == pyspark.sql.types.StructType:
    return pyspark.sql.types.StructType([fix_spark_schema(f) for f in schema.fields])
  if schema.__class__ == pyspark.sql.types.StructField:
    return pyspark.sql.types.StructField(schema.name, fix_spark_schema(schema.dataType), schema.nullable)
  if schema.__class__ == pyspark.sql.types.NullType:
    return pyspark.sql.types.StringType()
  return schema

【问题讨论】：

标签： mongodb apache-spark

【解决方案1】：

这是scala版本

  def deNullifyStruct(struct: StructType): StructType = {
    val items = struct.map { field => StructField(field.name, fixNullType(field.dataType), field.nullable, field.metadata) }
    StructType(items)
  }


  def fixNullType(dt: DataType): DataType = {
    dt match {
      case _: StructType => return deNullifyStruct(dt.asInstanceOf[StructType])
      case _: ArrayType =>
        val array = dt.asInstanceOf[ArrayType]
        return ArrayType(fixNullType(array.elementType), array.containsNull)
      case _: NullType => return StringType
      case _ => return dt
    }
  }

val df = // Initial df
// get new dataframe using modified schema
val newDF = spark.createDataFrame(df.rdd, deNullifyStruct(df.schema))

参考：Writing null values to Parquet in Spark when the NullType is inside a StructType

【讨论】：