代码之家 › 专栏 › 技术社区 › Maria Montenegro baudsp

在Spark Scala中将List<List<Long,Float,Float,Integer,Integer>>转换为Array<StructType>

user-defined-functions apache-spark scala

Maria Montenegro baudsp · 技术社区 · 6 年前

我想转换一个表示 List<List<Long,Float,Float,Integer,Integer>> 排成一列。为了实现这一点,我使用具有以下结构的UDF函数:

字符串的一个例子是 [[337, -115.0, -17.5, 6225, 189],[85075, -112.0, -12.5, 6225, 359]]

    def convertToListOfListComplex(ListOfList: String, regex: String): Array[StructType]
  ={
    val notBracket = ListOfList.dropRight(1).drop(1)
    val SplitString = notBracket.split("]").map(x=>if (x.startsWith("[")) x.drop(1) else x.drop(2))
    SplitString(0).replaceAll("\\s", "")

    val result =SplitString map {
      case s => {
        val split = s.replaceAll("\\s", "").trim.split(",")
        case class Row(a: Long, b: Float, c: Float, d: Int, e: Int)
        val element = Row(split(0).toLong, split(1).toFloat, split(2).toFloat, split(3).toInt, split(4).toInt)
        val schema = `valid code to transform to case class to StructType`
      }
    }
    return result
  }

我用的是Spark 2.2。我尝试了不同的解决方案,但在获取结构类型数组时发现问题,获取编译错误或执行失败。有什么建议吗??

1 回复 | 直到 6 年前

Ramesh Maharjan 6 年前

为了进行测试,我创建了一个测试数据框,其中的字符串在问题中提到

val df = Seq(
  Tuple1("[[337, -115.0, -17.5, 6225, 189],[85075, -112.0, -12.5, 6225, 359]]")
).toDF("col")

那就是

+-------------------------------------------------------------------+
|col                                                                |
+-------------------------------------------------------------------+
|[[337, -115.0, -17.5, 6225, 189],[85075, -112.0, -12.5, 6225, 359]]|
+-------------------------------------------------------------------+

root
 |-- col: string (nullable = true)

这个 udf 函数应如下所示

import org.apache.spark.sql.functions._
def convertToListOfListComplex = udf((ListOfList: String) => {
  ListOfList.split("],\\[")
    .map(x => x.replaceAll("[\\]\\[]", "").split(","))
    .map(splitted => rowTest(splitted(0).trim.toLong, splitted(1).trim.toFloat, splitted(2).trim.toFloat, splitted(3).trim.toInt, splitted(4).trim.toInt))
})

哪里 rowTest 是一个 case class 范围外定义 作为

case class rowTest(a: Long, b: Float, c: Float, d: Int, e: Int)

打电话给 自定义项 功能

df.withColumn("converted", convertToListOfListComplex(col("col")))

应该给你输出

+-------------------------------------------------------------------+--------------------------------------------------------------------+
|col                                                                |converted                                                           |
+-------------------------------------------------------------------+--------------------------------------------------------------------+
|[[337, -115.0, -17.5, 6225, 189],[85075, -112.0, -12.5, 6225, 359]]|[[337, -115.0, -17.5, 6225, 189], [85075, -112.0, -12.5, 6225, 359]]|
+-------------------------------------------------------------------+--------------------------------------------------------------------+


root
 |-- col: string (nullable = true)
 |-- converted: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- a: long (nullable = false)
 |    |    |-- b: float (nullable = false)
 |    |    |-- c: float (nullable = false)
 |    |    |-- d: integer (nullable = false)
 |    |    |-- e: integer (nullable = false)

站在更安全的一边 你可以用 Try/getOrElse 在 自定义项 作为

import org.apache.spark.sql.functions._
def convertToListOfListComplex = udf((ListOfList: String) => {
  ListOfList.split("],\\[")
    .map(x => x.replaceAll("[\\]\\[]", "").split(","))
    .map(splitted => rowTest(Try(splitted(0).trim.toLong).getOrElse(0L), Try(splitted(1).trim.toFloat).getOrElse(0F), Try(splitted(2).trim.toFloat).getOrElse(0F), Try(splitted(3).trim.toInt).getOrElse(0), Try(splitted(4).trim.toInt).getOrElse(0)))
})

我希望答案对你有帮助