Lindorm（HBase）数据入库与ETL

矢量数据入库

Lindorm（HBase）矢量数据导入，请参见快速入门。

栅格数据入库

1. Pipeline技术

   Pipeline模型是DLA Ganos基于GeoTrellis开源项目开发的用于栅格数据快速加载、处理和入库的ETL技术。

   Pipeline模型包含了一系列功能模块：如读取数据（Load），转换（Transform），保存数据（Save）等。DLA Ganos Pipeline模型一般表示为一个JSON对象，其主要对象称为pipeline，该对象是要执行的步骤数组（还有一些JSON对象，我们将其称为Stage Objects）。DLA Ganos整个入库操作的Pipeline流程与相关参数全部通过一个JSON对象进行定义，一个简单的JSON脚本如下所示：

   [
     {
       "uri" : "OSS资源URI",
       "type" : "singleband.spatial.read.oss"
     },
     {
       "resample_method" : "nearest-neighbor",
       "type" : "singleband.spatial.transform.tile-to-layout"
     },
     {
       "crs" : "EPSG:3857",
       "scheme" : {
         "crs" : "epsg:3857",
         "tileSize" : 256,
         "resolutionThreshold" : 0.1
       },
       "resample_method" : "nearest-neighbor",
       "type" : "singleband.spatial.transform.buffered-reproject"
     },
     {
       "end_zoom" : 0,
       "resample_method" : "nearest-neighbor",
       "type" : "singleband.spatial.transform.pyramid"
     },
     {
       "name" : "mask",
       "uri" : "oss://geotrellis-test/colingw/pipeline/",
       "key_index_method" : {
         "type" : "zorder"
       },
       "scheme" : {
         "crs" : "epsg:3857",
         "tileSize" : 256,
         "resolutionThreshold" : 0.1
       },
       "type" : "singleband.spatial.write"
     }
   ]

2. 入库流程
   1. 导入相关依赖。

      import geotrellis.layer._
      import geotrellis.spark.pipeline._
      import geotrellis.spark.pipeline.json._
      import geotrellis.spark._
      import geotrellis.spark.store.kryo.KryoRegistrator
      import org.apache.spark.{SparkConf, SparkContext}
      import scala.util.{Failure, Try}

   2. 初始化Spark环境。

        val conf =
          new SparkConf（）
            .setMaster（"local[*]"）
            .setAppName（"Spark Tiler"）
            .set（"spark.serializer", "org.apache.spark.serializer.KryoSerializer"）
            .set（"spark.kryo.registrator", classOf[KryoRegistrator].getName）
        conf.set（"spark.kryoserializer.buffer.max", "2047m"）
        implicit val sc = new SparkContext（conf）

   3. 定义Pipeline JSON描述。
      以下示例是一个简单的pipeline模型，该模型定义的操作如下：

      • 定义导入文件的URI与加载驱动。
      • 数据分块模式（tile-to-layout）。
      • 数据转换与冲投影等操作。
      • 数据写入地址（Lindorm）。

      该模型的详细配置如下所示：

       val pipeline: String =
          """
            |[
            |  {
            |    "uri" : "OSS资源URI",
            |    "time_tag" : "TIFFTAG_DATETIME",
            |    "time_format" : "yyyy:MM:dd HH:mm:ss",
            |    "type" : "singleband.spatial.read.hadoop"
            |  },
            |  {
            |    "resample_method" : "nearest-neighbor",
            |    "type" : "singleband.spatial.transform.tile-to-layout"
            |  },
            |  {
            |    "crs" : "EPSG:3857",
            |    "scheme" : {
            |      "crs" : "EPSG:3857",
            |      "tileSize" : 256,
            |      "resolutionThreshold":0.1
            |    },
            |    "resample_method" : "nearest-neighbor",
            |    "type" : "singleband.spatial.transform.buffered-reproject"
            |  },
            |  {
            |    "end_zoom" : 0,
            |    "resample_method" : "nearest-neighbor",
            |    "type" : "singleband.spatial.transform.pyramid"
            |  },
            |  {
            |    "name" : "srtm",
            |    "uri" : "hbase://localhost:2181?master=localhost&attributes=attributes&layers=srtm-tms-layers",
            |    "pyramid" : true,
            |    "key_index_method" : {
            |      "type" : "zorder"
            |    },
            |    "scheme" : {
            |      "tileCols" : 256,
            |      "tileRows" : 256
            |    },
            |    "type" : "singleband.spatial.write"
            |  }
            |]
          """.stripMargin

   4. 运行Pipeline模型。

      //首先解析JSON描述的Pipeline模型，生成表达式集合：
      val list: List[PipelineExpr] = pipeline.pipelineExpr match {
          case Right（r） => r
          case Left（e） => throw e
      }
      
      //执行pipeline模型：
      val erasedNode = list.erasedNode
        Try {
          erasedNode.eval[Stream[（Int, TileLayerRDD[SpatialKey]）]]
        } match {
          case Failure（e） => println（"run failed as expected"）; throw e
          case _ =>
        }

配置文件参考

{
   "uri" : "{oss| file | hdfs | ...}://...",
   "time_tag" : "TIFFTAG_DATETIME", // optional field
   "time_format" : "yyyy:MM:dd HH:mm:ss", // optional field
   "type" : "{singleband | multiband}.{spatial | temporal}.read.{oss | hadoop}"
}

{
   "name" : "layerName",
   "uri" : "{oss| file | hdfs | ...}://...",
   "key_index_method" : {
      "type" : "{zorder | hilbert}",
      "temporal_resolution": 1 // optional, if set - temporal index is used
   },
   "scheme" : {
      "crs" : "epsg:3857",
      "tileSize" : 256,
      "resolutionThreshold" : 0.1
   },
   "type" : "{singleband | multiband}.{spatial | temporal}.write"
}

关于Layout Scheme