Обрабатывать модели с циклическими ссылками в Spark SQL? - PullRequest
Новая
       8

Обрабатывать модели с циклическими ссылками в Spark SQL?

0 голосов
/ 22 сентября 2018

Вчера (практически полный журнал) Я попытался найти элегантный способ представления модели с круговыми ссылками в Scala / Spark SQL 2.2.1

Допустим, этооригинальный подход к модели, который, конечно, не работает (имейте в виду, что у реальной модели есть десятки атрибутов): 

case class Branch(id: Int, branches: List[Branch] = List.empty)
case class Tree(id: Int, branches: List[Branch])

val trees = Seq(Tree(1, List(Branch(2, List.empty), Branch(3, List(Branch(4, List.empty))))))

val ds = spark.createDataset(trees)
ds.show

И это ошибка, которую она выдает: 

java.lang.UnsupportedOperationException: cannot have circular references in class, but got the circular reference of class Branch

Я знаю, что максимальный уровень иерархии , который у нас есть, равен 5 .Итак, как обходной путь, я, хотя в чем-то вроде: 

case class BranchLevel5(id: Int)
case class BranchLevel4(id: Int, branches: List[BranchLevel5] = List.empty)
case class BranchLevel3(id: Int, branches: List[BranchLevel4] = List.empty)
case class BranchLevel2(id: Int, branches: List[BranchLevel3] = List.empty)
case class BranchLevel1(id: Int, branches: List[BranchLevel2] = List.empty)
case class Tree(id: Int, branches: List[BranchLevel1])

Конечно, это работает.Но это совсем не элегантно, и вы можете представить себе боль в реализации (удобочитаемость, связывание, обслуживание, удобство использования, дублирование кода и т. Д.)с круговыми ссылками в модели?

1 Ответ

0 голосов
/ 22 сентября 2018

Если вы в порядке с использованием частного API, то я нашел один способ, который просто работает: обрабатывать всю самоссылочную структуру как определяемый пользователем тип.Я следую подходу из этого ответа: https://stackoverflow.com/a/51957666/1823254.

package org.apache.spark.custom.udts // we're calling some private API so need to be under 'org.apache.spark'

import java.io._
import org.apache.spark.sql.types.{DataType, UDTRegistration, UserDefinedType}

class BranchUDT extends UserDefinedType[Branch] {

  override def sqlType: DataType = org.apache.spark.sql.types.BinaryType
  override def serialize(obj: Branch): Any = {
    val bos = new ByteArrayOutputStream()
    val oos = new ObjectOutputStream(bos)
    oos.writeObject(obj)
    bos.toByteArray
  }
  override def deserialize(datum: Any): Branch = {
    val bis = new ByteArrayInputStream(datum.asInstanceOf[Array[Byte]])
    val ois = new ObjectInputStream(bis)
    val obj = ois.readObject()
    obj.asInstanceOf[Branch]
  }

  override def userClass: Class[Branch] = classOf[Branch]
}

object BranchUDT {
  def register() = UDTRegistration.register(classOf[Branch].getName, classOf[BranchUDT].getName)
}

Просто создайте и зарегистрируйте пользовательский UDT, и все! 

BranchUDT.register()
val trees = Seq(Tree(1, List(Branch(2, List.empty), Branch(3, List(Branch(4, List.empty))))))

val ds = spark.createDataset(trees)
ds.show(false)

//+---+----------------------------------------------------+
//|id |branches                                            |
//+---+----------------------------------------------------+
//|1  |[Branch(2,List()), Branch(3,List(Branch(4,List())))]|
//+---+----------------------------------------------------+
...