1

在使用 Geomesa 和 Scala 时，我一直在尝试使用以下代码片段对 Spark Dataframe 中的 2 列进行编码，但我不断收到一个问题，Scala 似乎无法将返回的对象序列化为 Dataframe。使用 Postgres 和 PostGIS 时，生活很轻松 - 这是一个简单的问题，还是有更好的库可以处理来自包含双格式纬度和经度的 Spark Dataframe 的地理空间查询？

我在 SBT 中使用的版本是：

  • 火花：2.3.0
  • 斯卡拉：2.11.12
  • 几何图形：2.2.1
  • jst-*: 1.17.0-SNAPSHOT

线程“主”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)
4

2 回答 2

2

如果您不使用底层 GeoMesa 存储将数据加载到 spark 会话中，则需要显式注册 JTS 类型：

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

这将注册ST_操作以及 JTS 编码器。

于 2019-03-07T00:57:47.953 回答
1

用简单的英语，例外是说：

我不知道如何将 Point 转换为 Spark 类型。

如果您在数据集中将纬度和经度保持为双倍，那么您应该没问题，但是一旦您使用像 Point 这样的对象，您就需要告诉 Spark 如何转换它。在 Spark 术语中，这些称为编码器，您可以创建自定义编码器。

或者你切换到一个不需要转换的 RDD，只要你不介意丢失 Spark SQL 的东西。

于 2019-03-06T23:13:14.597 回答