LayerToGeotiff / MultibandLayerToGeotiff use foreachPartition with se…

…rial hadoopConfiguration
biggis-project · Mar 8, 2018 · b50a234 · b50a234
1 parent 9bd5f16
commit b50a234
Show file tree

Hide file tree

Showing 2 changed files with 54 additions and 25 deletions.
diff --git a/src/main/scala/biggis/landuse/spark/examples/LayerToGeotiff.scala b/src/main/scala/biggis/landuse/spark/examples/LayerToGeotiff.scala
@@ -97,20 +97,34 @@ object LayerToGeotiff extends LazyLogging {
       //.tileToLayout(metadata.cellType, metadata.layout, Utils.RESAMPLING_METHOD)
       //.repartition(Utils.RDD_PARTITIONS)
 
-      /*
-      outputRdd.foreachPartition{ partition =>
-        partition.map(_.write(new Path("hdfs://..."), serConf.value))
-      } // */
-      outputRdd.foreach(mbtile => {
-        val (key, tile) = mbtile
-        val (col, row) = (key.col, key.row)
-        val tileextent: Extent = metadata.layout.mapTransform(key)
-        val filename = new Path(outputPath + "_" + col + "_" + row + ".tif")
-        logger info s" writing: '${filename.toString}'"
-        GeoTiff(tile, tileextent, crs)
-          .write(filename, serConf.value)
+      val useSerializedHadoopConfig = true
+      if(useSerializedHadoopConfig){
+        // ToDo: test Spark Cluster version
+        outputRdd.foreachPartition { partition =>
+          partition.foreach { tuple =>
+            val (key, tile) = tuple
+            val (col, row) = (key.col, key.row)
+            val tileextent: Extent = metadata.layout.mapTransform(key)
+            val filename = new Path(outputPath + "_" + col + "_" + row + ".tif")
+            logger info s" writing: '${filename.toString}'"
+            GeoTiff(tile, tileextent, crs)
+              .write(filename, serConf.value)
+          }
+        }
+      } else {
+        // only for local debugging - do not use in cloud // ToDo: delete after testing
+        outputRdd.foreach(mbtile => {
+          val (key, tile) = mbtile
+          val (col, row) = (key.col, key.row)
+          val tileextent: Extent = metadata.layout.mapTransform(key)
+          //val filename = new Path(outputPath + "_" + col + "_" + row + ".tif")
+          //logger info s" writing: '${filename.toString}'"
+          GeoTiff(tile, tileextent, crs)
+            //.write(filename.toString) //.write(filename, serConf.value)
+            .write(outputPath + "_" + col + "_" + row + ".tif")
+        }
+        )
       }
-      )
     }
 
     //sc.stop()

diff --git a/src/main/scala/biggis/landuse/spark/examples/MultibandLayerToGeotiff.scala b/src/main/scala/biggis/landuse/spark/examples/MultibandLayerToGeotiff.scala
@@ -153,19 +153,34 @@ object MultibandLayerToGeotiff extends LazyLogging{
       //.repartition(myRDD_PARTITIONS)
       //.tileToLayout(myMetadata.cellType, myMetadata.layout, myRESAMPLING_METHOD)
 
-      /*
-      outputRdd.foreachPartition{ partition =>
-        partition.map(_.write(new Path("hdfs://..."), serConf.value))
-      } // */
-      outputRdd.foreach(mbtile => {
-        val (key, tile) = mbtile
-        val (col, row) = (key.col, key.row)
-        val tileextent: Extent = metadata.layout.mapTransform(key)
-        val filename = new Path(outputPath + "_" + col + "_" + row + ".tif")
-        MultibandGeoTiff(tile, tileextent, crs)
-          .write(filename, serConf.value)
+      val useSerializedHadoopConfig = true
+      if (useSerializedHadoopConfig) {
+        // ToDo: test Spark Cluster version
+        outputRdd.foreachPartition { partition =>
+          partition.foreach { tuple =>
+            val (key, tile) = tuple
+            val (col, row) = (key.col, key.row)
+            val tileextent: Extent = metadata.layout.mapTransform(key)
+            val filename = new Path(outputPath + "_" + col + "_" + row + ".tif")
+            logger info s" writing: '${filename.toString}'"
+            MultibandGeoTiff(tile, tileextent, crs)
+              .write(filename, serConf.value)
+          }
+        }
+      } else {
+        // only for local debugging - do not use in cloud // ToDo: delete after testing
+        outputRdd.foreach(mbtile => {
+          val (key, tile) = mbtile
+          val (col, row) = (key.col, key.row)
+          val tileextent: Extent = metadata.layout.mapTransform(key)
+          //val filename = new Path(outputPath + "_" + col + "_" + row + ".tif")
+          //logger info s" writing: '${filename.toString}'"
+          MultibandGeoTiff(tile, tileextent, crs)
+            //.write(filename.toString) //.write(filename, serConf.value)
+            .write(outputPath + "_" + col + "_" + row + ".tif")
+        }
+        )
       }
-      )
     }
 
     ////val raster: Raster[MultibandTile] = tile.reproject(metadata.extent, metadata.crs, metadata.crs)