Блог

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

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

• spark: 2.3.0
• scala: 2.11.12
• geomesa: 2.2.1
• jst-*: 1.17.0-СНИМОК

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

 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)
  

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

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

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

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

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

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

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