Как создать TypedColumn в наборе Spark Dataset и управлять им?

Question

Я пытаюсь выполнить агрегацию с использованием mapGroups, которая возвращает SparseMatrix как один из столбцов и суммирует столбцы.

Я создал схему для отображаемых строк, чтобы указать имена столбцов. Набирается матричный столбец org.apache.spark.mllib.linalg.Matrix. Если я не запускаю toDF перед выполнением агрегации (select(sum("mycolumn")), я получаю ошибку несоответствия типа (required: org.apache.spark.sql.TypedColumn[MySchema,?]). Если я включаю toDF, я получаю ошибку несоответствия типа: cannot resolve 'sum(mycolumn)' due to data type mismatch: function sum requires numeric types, not org.apache.spark.mllib.linalg.MatrixUDT. Итак, каков правильный способ сделать это?

Answer 1

Похоже, вы боретесь, по крайней мере, с двумя различными проблемами. Давайте предположим, что вы имеете Dataset так:

val ds = Seq(
    ("foo", Matrices.dense(3, 2, Array(1.0, 3.0, 5.0, 2.0, 4.0, 6.0))), 
    ("foo", Matrices.dense(3, 2, Array(1.0, 3.0, 5.0, 2.0, 4.0, 6.0))) 
).toDS 

Выбор TypedColumn:

• с помощью неявного преобразования с $:

  ds.select(col("_1").as[String]) 
  

• использованием o.a.s.sql.functions.col:

  ds.select(col("_1").as[String]) 
  

Добавление матрицы:

• MLLib Matrix и MatrixUDT не реализуют дополнение. Это означает, что вы не сможете sum функции или уменьшить с +
• вы можете использовать третью партийную библиотеку линейной алгебры, но это не поддерживается в Спарк SQL/Спарк Dataset

Если вы действительно хотите сделать это с Datsets вы можете попробовать сделать что-то вроде этого:

ds.groupByKey(_._1).mapGroups(
    (key, values) => { 
    val matrices = values.map(_._2.toArray) 
    val first = matrices.next 
    val sum = matrices.foldLeft(first)(
     (acc, m) => acc.zip(m).map { case (x, y) => x + y } 
    ) 
    (key, sum) 
}) 

карту обратно к матрицам, но лично я бы просто преобразовать в РДУ и использовать breeze.