Преобразовать столбец Dataframe String в Array [Int]

Question

Я новичок в Scala и Spark и пытаюсь прочитать файл csv локально (для тестирования):

val spark = org.apache.spark.sql.SparkSession.builder.master("local").appName("Spark CSV Reader").getOrCreate;
val topics_df = spark.read.format("csv").option("header", "true").load("path-to-file.csv")
topics_df.show(10)

Вот как выглядит файл:

+-----+--------------------+--------------------+
|topic|         termindices|         termweights|
+-----+--------------------+--------------------+
|   15|[21,31,51,108,101...|[0.0987100701,0.0...|
|   16|[42,25,121,132,55...|[0.0405490884,0.0...|
|    7|[1,23,38,7,63,0,1...|[0.1793091892,0.0...|
|    8|[13,40,35,104,153...|[0.0737646511,0.0...|
|    9|[2,10,93,9,158,18...|[0.1639456608,0.1...|
|    0|[28,39,71,46,123,...|[0.0867449145,0.0...|
|    1|[11,34,36,110,112...|[0.0729913664,0.0...|
|   17|[6,4,14,82,157,61...|[0.1583892199,0.1...|
|   18|[9,27,74,103,166,...|[0.0633899386,0.0...|
|   19|[15,81,289,218,34...|[0.1348582482,0.0...|
+-----+--------------------+--------------------+

с

ReadSchema: struct<topic:string,termindices:string,termweights:string>

Предполагается, что столбец termindices имеет тип Array[Int], но при сохранении в CSV это String (обычно это не будет проблемой при извлечении из баз данных).

Как преобразовать тип и в конечном итоге привести DataFrame к:

case class TopicDFRow(topic: Int, termIndices: Array[Int], termWeights: Array[Double])

У меня есть функция, готовая выполнить преобразование:

termIndices.substring(1, termIndices.length - 1).split(",").map(_.toInt)

IЯ посмотрел на udf и несколько других решений, но я убежден, что должен быть намного более чистый и быстрый способ выполнить указанное преобразование.Любая помощь с благодарностью!

Answer 1

UDF следует избегать, когда возможно использовать более эффективные встроенные функции Spark.Насколько мне известно, нет лучшего способа, чем предложенный;удалить первый и последний символы строки, разделить и преобразовать.

Используя встроенные функции, это можно сделать следующим образом:

df.withColumn("termindices", split($"termindices".substr(lit(2), length($"termindices")-2), ",").cast("array<int>"))
  .withColumn("termweights", split($"termweights".substr(lit(2), length($"termweights")-2), ",").cast("array<double>"))
  .as[TopicDFRow]

substr если на основе 1-индекса, поэтому для удаления первого символа мы начинаем с 2. Второйаргумент - это длина, которую нужно взять (не конечная точка), следовательно, -2.

Последняя команда преобразует кадр данных в набор данных типа TopicDFRow.