Вычислить максимальную длину, назначенную каждому элементу, используя scala

Question

Например, это содержимое файла:

20,1,helloworld,alaaa
2,3,world,neww
1,223,ala,12341234

Требуемый вывод "

0-> 2
1-> 3
2-> 10
3-> 8

Я хочу найти максимальную длину, назначенную каждому элементу.

Answer 1

Я бы порекомендовал использовать метод U,combOp:(U,U)=>U)(implicitevidence$30:scala.reflect.ClassTag[U]):U" rel="nofollow noreferrer"> агрегирования СДР:

val rdd = sc.textFile("/path/to/textfile").
  map(_.split(","))
// res1: Array[Array[String]] = Array(
//   Array(20, 1, helloworld, alaaa), Array(2, 3, world, neww), Array(1, 223, ala, 12341234)
// )

val seqOp = (m: Array[Int], r: Array[String]) =>
  (r zip m).map( t => Seq(t._1.length, t._2).max )

val combOp = (m1: Array[Int], m2: Array[Int]) =>
  (m1 zip m2).map( t => Seq(t._1, t._2).max )

val size = rdd.collect.head.size

rdd.
  aggregate( Array.fill[Int](size)(0) )( seqOp, combOp ).
  zipWithIndex.map(_.swap).
  toMap
// res2: scala.collection.immutable.Map[Int,Int] = Map(0 -> 2, 1 -> 3, 2 -> 10, 3 -> 8)

Обратите внимание, что aggregate принимает:

1. массив из 0 (размером равнымк размеру строки rdd) в качестве начального значения
2. функция seqOp для вычисления максимальной длины строки в пределах раздела и
3. другая функция combOp для объединения результатов по разделам дляокончательные максимальные значения.

Answer 2

Возможно расширить это до любое количество столбцов .Сначала прочитайте файл как фрейм данных:

val df = spark.read.csv("path")

Затем создайте выражение SQL для каждого столбца и оцените его с помощью expr:

val cols = df.columns.map(c => s"max(length(cast($c as String)))").map(expr(_))

Выберите новые столбцы как массив искрыто Map:

df.select(array(cols:_*)).as[Seq[Int]].collect()
  .head
  .zipWithIndex.map(_.swap)
  .toMap

Это должно дать вам желаемый Map.

Map(0 -> 2, 1 -> 3, 2 -> 10, 3 -> 8)

Answer 3

Обновление:

1. В примере OP предполагается, что они будут одинаковой длины.

2. Использование Spark-SQL и max (length ()) в столбцах DF - идея, предложенная в этом ответе.

Вы можете сделать:

val xx = Seq(
             ("20","1","helloworld","alaaa"),
             ("2","3","world","neww"),
             ("1","223","ala","12341234")
            ).toDF("a", "b", "c", "d")

xx.registerTempTable("yy")

spark.sql("select max(length(a)), max(length(b)), max(length(c)), max(length(d)) from yy")