带你入门GeoSpark系列之一【环境篇】

2021-05-06

GeoSpark是基于Spark之上的分布式群集计算系统。GeoSpark扩展了Spark Core和SparkSQL并提出了空间弹性分布式数据集（Spatial Resilient Distributed Datasets (SRDDs)）同时提供了可视化组件。简而言之就是可以利用它在Spark上做空间运算。能够基于经纬度等信息创建点（Point）线(LineString）面(Polygon)。并提供了几种空间查询:空间临近查询(Spatial KNN Query)、空间范围查询( Spatial Range Query)、空间连接查询(Spatial Join Query)和距离连接查询(Distance Join Query)。

前言

由于项目需要处理一些地理空间数据，而原生Spark不能满足，这几天捣鼓GeoSpark有一些收获，开个坑总结一下。

GeoSpark介绍

GeoSpark是基于Spark之上的分布式群集计算系统。GeoSpark扩展了Spark Core和SparkSQL并提出了空间弹性分布式数据集（Spatial Resilient Distributed Datasets (SRDDs)）同时提供了可视化组件。简而言之就是可以利用它在Spark上做空间运算。能够基于经纬度等信息创建点（Point）线(LineString）面(Polygon)。并提供了几种空间查询:空间临近查询(Spatial KNN Query)、空间范围查询( Spatial Range Query)、空间连接查询(Spatial Join Query)和距离连接查询(Distance Join Query)。

环境准备

JDK 1.8
Scala 2.11.x

Pom依赖

Idea创建简单的scala maven项目即可，然后配置Pom依赖
注意，GeoSpark的版本一定要和相应的Spark版本对应，怎么对应呢？
官方给出了一份说明:
https://datasystemslab.github.io/GeoSpark/download/GeoSpark-All-Modules-Maven-Central-Coordinates/

可以看到不同的spark对应的geospark-sql的artifactId后缀版本不一样!
同时我们也可以看出官方对spark的支持最新是到spark 2.3，所以避免踩坑笔者用了这个版本，其他版本未测试。

<properties>
    <scala.version>2.11.8</scala.version>
    <spark.version>2.3.4</spark.version>
    <scala.binary.version>2.11</scala.binary.version>
    <geospark.version>1.3.0</geospark.version>
</properties>

<dependencies>
    <dependency>
      <groupId>org.scala-lang</groupId>
      <artifactId>scala-library</artifactId>
      <version>${scala.version}</version>
    </dependency>

    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-core_${scala.binary.version}</artifactId>
      <version>${spark.version}</version>
    </dependency>
    
    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-sql_${scala.binary.version}</artifactId>
      <version>${spark.version}</version>
    </dependency>
    
    <dependency>
      <groupId>org.datasyslab</groupId>
      <artifactId>geospark</artifactId>
      <version>${geospark.version}</version>
    </dependency>
    
    <dependency>
      <groupId>org.datasyslab</groupId>
      <artifactId>geospark-sql_2.3</artifactId>
      <version>${geospark.version}</version>
    </dependency>

  </dependencies>

尝鲜

新建一个CSV文件checkin.csv：

1
2
3

-88.175933,32.360763,gas
-88.388954,32.357073,bar
-88.221102,32.35078,restaurant

Code:

package com.suddev.bigdata.core

import org.apache.spark.serializer.KryoSerializer
import org.apache.spark.{SparkConf, SparkContext}
import org.datasyslab.geospark.enums.FileDataSplitter
import org.datasyslab.geospark.serde.GeoSparkKryoRegistrator
import org.datasyslab.geospark.spatialRDD.PointRDD


object DemoApp {
  def main(args: Array[String]): Unit = {
    // 创建SparkConf
    val conf = new SparkConf().
      setAppName("GeoSparkDemo1").
      setMaster("local[*]").
      set("spark.serializer", classOf[KryoSerializer].getName).
      set("spark.kryo.registrator", classOf[GeoSparkKryoRegistrator].getName)
    val sc = new SparkContext(conf)

    val pointRDDInputLocation = "data/checkin.csv"
    // 这个变量控制我们的地理经度和纬度在数据的哪两列，我们这里是第0,1列，Offset就设置为0
    val pointRDDOffset = 0
    val pointRDDSplitter = FileDataSplitter.CSV
    // 这个参数允许我们除了经纬度外还可以携带其他自定义数据
    val carryOtherAttributes = true
    val objectRDD = new PointRDD(sc, pointRDDInputLocation,pointRDDOffset, pointRDDSplitter, carryOtherAttributes)
    // 获取rawRDD进行遍历输出
    objectRDD.rawSpatialRDD.rdd.collect().foreach(println)

  }
}

Output:

out