Не найден кодировщик для org.locationtech.jts.geom.Point - PullRequest
Новая
       38

Не найден кодировщик для org.locationtech.jts.geom.Point

0 голосов
/ 07 марта 2019

При использовании Geomesa и Scala я пытался закодировать 2 столбца в кадре данных Spark, используя приведенные ниже фрагменты, но я постоянно получаю сообщение о том, что Scala не может сериализовать возвращенные объекты в кадр данных. При использовании Postgres и PostGIS жизнь проста - это простой вопрос или есть лучшая библиотека, которая может обрабатывать геопространственные запросы, поступающие из кадра данных Spark, который содержит широту и долготу в двойном формате?

Версии, которые я использую в моем SBT:

  • искра: 2.3.0
  • Скала: 2.11.12
  • Геомеза: 2.2.1
  • jst- *: 1.17.0-SNAPSHOT

Исключение в потоке "main" java.lang.UnsupportedOperationException: не найден кодировщик для org.locationtech.jts.geom.Point 

import org.apache.spark.sql.SparkSession
import org.locationtech.jts.geom.{Coordinate, GeometryFactory}
import org.apache.spark.sql.functions.col

import org.apache.spark.sql.types._
import org.locationtech.geomesa.spark.jts._


object GetRandomData {


  def main(sysArgs: Array[String]) {

    @transient val spark: SparkSession = {
      SparkSession
        .builder()
        .config("spark.ui.enabled", "false")
        .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
        .config("spark.kryoserializer.buffer.mb","24")
        .appName("GetRandomData")
        .master("local[*]")
        .getOrCreate()
    }
    val sc = spark.sparkContext
    sc.setLogLevel("ERROR")
    import spark.sqlContext.implicits._

    var coordinates = sc.parallelize(
      List(
        (35.40466, -80.905458),
        (35.344079, -80.872267),
        (35.139606, -80.840845),
        (35.537786, -80.780051),
        (35.525361, -83.031932),
        (34.928323, -80.766732),
        (35.533865, -82.72344),
        (35.50997,  -80.588572),
        (35.286251, -83.150514),
        (35.558519, -81.067069),
        (35.569311, -80.916993),
        (35.835867, -81.067904),
        (35.221695, -82.662141)
      )
    ).
    toDS().
    toDF("geo_lat", "geo_lng")

    coordinates = coordinates.select(coordinates.columns.map(c => col(c).cast(DoubleType)) : _*)
    coordinates.show()
    val testing = coordinates.map(r => new GeometryFactory().createPoint(new Coordinate(3.4, 5.6)))
    val coordinatesPointDf = coordinates.withColumn("point", st_makePoint(col("geo_lat"), col("geo_lng")))

  }
}

Исключение составляет: 

    Exception in thread "main" java.lang.UnsupportedOperationException: No Encoder found for org.locationtech.jts.geom.Point
- root class: "org.locationtech.jts.geom.Point"
  at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$org$apache$spark$sql$catalyst$ScalaReflection$$serializerFor$1.apply(ScalaReflection.scala:643)
  at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$org$apache$spark$sql$catalyst$ScalaReflection$$serializerFor$1.apply(ScalaReflection.scala:445)
  at scala.reflect.internal.tpe.TypeConstraints$UndoLog.undo(TypeConstraints.scala:56)
  at org.apache.spark.sql.catalyst.ScalaReflection$class.cleanUpReflectionObjects(ScalaReflection.scala:824)
  at org.apache.spark.sql.catalyst.ScalaReflection$.cleanUpReflectionObjects(ScalaReflection.scala:39)
  at org.apache.spark.sql.catalyst.ScalaReflection$.org$apache$spark$sql$catalyst$ScalaReflection$$serializerFor(ScalaReflection.scala:445)
  at org.apache.spark.sql.catalyst.ScalaReflection$.serializerFor(ScalaReflection.scala:434)
  at org.apache.spark.sql.catalyst.encoders.ExpressionEncoder$.apply(ExpressionEncoder.scala:71)
  at org.locationtech.geomesa.spark.jts.encoders.SpatialEncoders$class.jtsPointEncoder(SpatialEncoders.scala:21)
  at org.locationtech.geomesa.spark.jts.package$.jtsPointEncoder(package.scala:17)
  at GetRandomData$.main(Main.scala:50)
  at GetRandomData.main(Main.scala)

Ответы [ 2 ]

2 голосов
/ 07 марта 2019

Если вы не используете базовое хранилище GeoMesa для загрузки данных в сеанс искры, вам необходимо явно зарегистрировать типы JTS с помощью: 

org.apache.spark.sql.SQLTypes.init(spark.sqlContext)

Это зарегистрирует операции ST_ кака также кодеры JTS.

1 голос
/ 07 марта 2019

На простом английском языке исключение говорит:

Я не знаю, как преобразовать Point в тип Spark.

Если вы сохраняете широту и долготу в вашем наборе данных как двойные, тогда у вас все будет хорошо, но как только вы используете объект, такой как Point, вам нужно будет указать Spark, как преобразовать его. В терминах Spark они называются кодировщиками, и вы можете создавать собственные.

Или вы переключаетесь на RDD, где преобразование не требуется, если вы не против потерять Spark SQL.

...