模式匹配的 Scala 迭代器

Question

我需要帮助来使用 DataFrame 迭代这段用 Spark-Scala 编写的代码。我是 Scala 的新手，所以如果我的问题看起来微不足道，我深表歉意。

函数非常简单：给定一个dataframe，如果存在模式匹配，函数将转换列，否则select所有字段。

/* Load sources */
val df = sqlContext.sql("select id_vehicle, id_size, id_country, id_time from " + working_database + carPark);


val df2 = df.select(
       df.columns.map {
         case id_vehicle @ "id_vehicle" => df(id_vehicle).cast("Int").as(id_vehicle)
         case other         => df(other)
       }: _*
    )

这个函数，加上模式匹配，完美运行！

现在我有一个问题：有什么办法可以"iterate"这个吗？在实践中，我需要一个给定 dataframe、列的 Array[String] 的函数（column_1、column_2、...）和 type（int、double、float，...）的另一个 Array[String]，return 对我来说是相同的 dataframe，正确的转换为正确的位置。

我需要帮助:)

Answer 1

//Your supplied code fits nicely into this function
def castOnce(df: DataFrame, colName: String, typeName: String): DataFrame = {
    val colsCasted = df.columns.map{
        case colName => df(colName).cast(typeName).as(colName)
        case other => df(other)
    }
    df.select(colsCasted:_ *)
}

def castMany(df: DataFrame, colNames: Array[String], typeNames: Array[String]): DataFrame = {

   assert(colNames.length == typeNames.length, "The lengths are different")
   val colsWithTypes: Array[(String, String)] = colNames.zip(typeNames)
   colsWithTypes.foldLeft(df)((cAndType, newDf) => castOnce(newDf, cAndType._1, cAndType._2))
}

当您有一个函数，您只需要多次将其应用于同一事物时，折叠通常就是您想要的。上面的代码将两个数组压缩在一起，将它们合二为一。然后它遍历此列表，每次将您的函数应用于数据框，然后将下一对应用于结果数据框等。

根据您的修改，我填写了上面的函数。我没有编译器，所以我不能 100% 确定它是正确的。把它写出来后，我也对我原来的方法提出了质疑。以下是我认为更好的方法，但我将上一个留作参考。

def(df: DataFrame, colNames: Array[String], typeNames: Array[String]): DataFrame = {
    assert(colNames.length == typeNames.length, "The lengths are different")
    val nameToType: Map[String, String] = colNames.zip(typeNames).toMap
    val newCols= df.columns.map{dfCol =>
        nameToType.get(dfCol).map{newType => 
            df(dfCol).cast(newType).as(dfCol)
        }.getOrElse(df(dfCol))
    }
    df.select(newCols:_ *)
}

以上代码创建了一个列名到所需类型的映射。然后，对于数据框中的每一列，它会在地图中查找类型。如果该类型存在，我们将该列转换为该新类型。如果地图中不存在该列，那么我们默认直接使用 DataFrame 中的列。

然后我们 select 来自 DataFrame 的这些列

模式匹配的 Scala 迭代器

Scala iterator on pattern match

scala

pattern-matching

spark-dataframe