时间戳

首页 > 解决方案 > 通过对存储在列表中的字段宽度值进行切片/拆分，将 apache-spark 数据帧字符串列拆分为多列

scala - 通过对存储在列表中的字段宽度值进行切片/拆分，将 apache-spark 数据帧字符串列拆分为多列

问题描述

我有一个看起来像这样的数据框

+--------------------
|       unparsed_data|
+--------------------
|02020sometext5002...
|02020sometext6682...

我需要把它分成这样的东西

+--------------------
|fips  | Name     | Id ...    
+--------------------
|02020 | sometext | 5002...
|02020 | sometext | 6682...

我有一个这样的清单

val fields = List(
  ("fips", 5),
  (“Name”, 8),
  (“Id”, 27)
  ....more fields
)

我需要吐痰将前 5 个字符unparsed_data映射到fips，将接下来的 8 个字符unparsed_data映射到Name，然后将接下来的 27 个字符映射到Id，依此类推。我需要拆分使用/引用列表中提供的文件长度来进行拆分/切片，因为有很多字段并且unparsed_data字段很长。

我的 scala 仍然很漂亮，我认为答案看起来像这样

df.withColumn("temp_field", split("unparsed_data", //some regex created from the list values?)).map(i => //some mapping to the field names in the list)

非常感谢任何建议/想法

标签： scalaapache-sparkapache-spark-sql

解决方案

您可以使用foldLeft遍历您的列表，以使用substringfields从原始 DataFrame 迭代地创建列 。无论列表大小如何，它都适用：fields

import org.apache.spark.sql.functions._

val df = Seq(
  ("02020sometext5002"),
  ("03030othrtext6003"),
  ("04040moretext7004")
).toDF("unparsed_data")

val fields = List(
  ("fips", 5),
  ("name", 8),
  ("id", 4)
)

val resultDF = fields.foldLeft( (df, 1) ){ (acc, field) =>
    val newDF = acc._1.withColumn(
      field._1, substring($"unparsed_data", acc._2, field._2)
    )
    (newDF, acc._2 + field._2)
  }._1.
  drop("unparsed_data")

resultDF.show
// +-----+--------+----+
// | fips|    name|  id|
// +-----+--------+----+
// |02020|sometext|5002|
// |03030|othrtext|6003|
// |04040|moretext|7004|
// +-----+--------+----+

请注意， aTuple2[DataFrame, Int]用作累加器，foldLeft用于携带迭代变换的 DataFrame 和 的下一个偏移位置substring。

推荐阅读