Задача не сериализуется при перемещении простой функции из UDF

Question

Я пытаюсь выделить функции в UDF для написания тестов, однако это вызывает ошибки типа сериализации?Я пробовал разные форматы создания UDF, но он все еще не работает.

Почему это работает?

   protected val normalizeDomain = udf {(dealer_url: String) => 
     var domain = if (dealer_url.startsWith("http://") || dealer_url.startsWith("https://")) dealer_url else s"http://$dealer_url"
     domain = new URL(domain).getHost
     if (domain.startsWith("www.")) domain.replace("www.", "") else domain
   }

Но не это?

  def normalizeDomainDef(dealer_url: String):String = {
    var domain = if (dealer_url.startsWith("http://") || dealer_url.startsWith("https://")) dealer_url else s"http://$dealer_url"
    domain = new URL(domain).getHost
    if (domain.startsWith("www.")) domain.replace("www.", "") else domain
  }
  val normalizeDomain = udf[String, String](normalizeDomainDef)

При выполнении последнегоЯ получаю:

Caused by: java.io.NotSerializableException: line9e1150bafcc941e3b83a5f4ad173d2ae122.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$DealerVDPUrlMaker$
Serialization stack:
    - object not serializable (class: line9e1150bafcc941e3b83a5f4ad173d2ae122.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$DealerVDPUrlMaker$, value: line9e1150bafcc941e3b83a5f4ad173d2ae122.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$DealerVDPUrlMaker$@131552b)
    - field (class: line9e1150bafcc941e3b83a5f4ad173d2ae122.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$DealerVDPUrlMaker$$anonfun$3, name: $outer, type: class line9e1150bafcc941e3b83a5f4ad173d2ae122.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$DealerVDPUrlMaker$)
    - object (class line9e1150bafcc941e3b83a5f4ad173d2ae122.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$DealerVDPUrlMaker$$anonfun$3, <function1>)
    - element of array (index: 4)
    - array (class [Ljava.lang.Object;, size 5)
    - field (class: org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$10, name: references$1, type: class [Ljava.lang.Object;)

Answer 1

Скорее всего, проблема заключается в том, где находится ваш normalizeDomainDef. Когда вы используете def, это означает, что вы помещаете его в некоторый класс, и теперь весь экземпляр этого класса должен быть сериализован для вызова этого метода ивероятно, этот класс включает в себя множество несериализуемых вещей.Чтобы исправить это, попробуйте поместить функцию в независимый верхний уровень object, например:

// top level object
object MyUdfs {
  def normalizeDomainDef(dealer_url: String):String = {
    var domain = if (dealer_url.startsWith("http://") || dealer_url.startsWith("https://")) dealer_url else s"http://$dealer_url"
    domain = new URL(domain).getHost
    if (domain.startsWith("www.")) domain.replace("www.", "") else domain
  }

  val normalizeDomain = udf[String, String](normalizeDomainDef)
}