babylon/src/main/java/org/datasyslab/babylon/core/ImageGenerator.java - sedona - Git at Google

 /**
  * FILE: ImageGenerator.java
  * PATH: org.datasyslab.babylon.core.ImageGenerator.java
  * Copyright (c) 2015-2017 GeoSpark Development Team
  * All rights reserved.
  */
 package org.datasyslab.babylon.core;

 import java.awt.image.BufferedImage;

 import java.io.*;
 import java.net.URI;
 import java.util.List;

 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.log4j.Logger;
 import org.apache.spark.api.java.JavaPairRDD;
 import org.apache.spark.api.java.JavaRDD;
 import org.apache.spark.api.java.function.Function;
 import org.apache.spark.api.java.function.VoidFunction;

 import org.datasyslab.babylon.utils.ImageType;
 import org.datasyslab.babylon.utils.RasterizationUtils;
 import org.datasyslab.babylon.utils.S3Operator;
 import scala.Tuple2;

 import javax.imageio.ImageIO;


 // TODO: Auto-generated Javadoc
 /**
  * The Class ImageGenerator.
  */
 public class ImageGenerator implements Serializable{

 	/** The Constant logger. */
 	final static Logger logger = Logger.getLogger(ImageGenerator.class);

 	/**
 	 * Save raster image as local file.
 	 *
 	 * @param distributedImage the distributed image
 	 * @param outputPath the output path
 	 * @param imageType the image type
 	 * @param zoomLevel the zoom level
 	 * @param partitionOnX the partition on X
 	 * @param partitionOnY the partition on Y
 	 * @return true, if successful
 	 * @throws Exception the exception
 	 */
 	public boolean SaveRasterImageAsLocalFile(JavaPairRDD<Integer,ImageSerializableWrapper> distributedImage, final String outputPath, final ImageType imageType, final int zoomLevel, final int partitionOnX, final int partitionOnY) throws Exception
 	{
 		logger.info("[Babylon][SaveRasterImageAsLocalFile][Start]");
 		for(int i=0;i<partitionOnX*partitionOnY;i++) {
 			deleteLocalFile(outputPath+"-"+ RasterizationUtils.getImageTileName(zoomLevel,partitionOnX, partitionOnY,i),imageType);
 		}
 		distributedImage.foreach(new VoidFunction<Tuple2<Integer, ImageSerializableWrapper>>() {
 			@Override
 			public void call(Tuple2<Integer, ImageSerializableWrapper> integerImageSerializableWrapperTuple2) throws Exception {
 				SaveRasterImageAsLocalFile(integerImageSerializableWrapperTuple2._2.image, outputPath+"-"+RasterizationUtils.getImageTileName(zoomLevel,partitionOnX, partitionOnY,integerImageSerializableWrapperTuple2._1), imageType);
 			}
 		});
 		logger.info("[Babylon][SaveRasterImageAsLocalFile][Stop]");
 		return true;
 	}

 	/**
 	 * Save raster image as local file.
 	 *
 	 * @param distributedImage the distributed image
 	 * @param outputPath the output path
 	 * @param imageType the image type
 	 * @return true, if successful
 	 * @throws Exception the exception
 	 */
 	public boolean SaveRasterImageAsLocalFile(JavaPairRDD<Integer,ImageSerializableWrapper> distributedImage, String outputPath, ImageType imageType) throws Exception
 	{
 		List<Tuple2<Integer,ImageSerializableWrapper>> imagePartitions = distributedImage.collect();
 		for(Tuple2<Integer,ImageSerializableWrapper> imagePartition:imagePartitions)
 		{
 			SaveRasterImageAsLocalFile(imagePartition._2.image, outputPath+"-"+imagePartition._1, imageType);
 		}
 		return true;
 	}

 	/**
 	 * Save raster image as hadoop file.
 	 *
 	 * @param distributedImage the distributed image
 	 * @param outputPath the output path
 	 * @param imageType the image type
 	 * @param zoomLevel the zoom level
 	 * @param partitionOnX the partition on X
 	 * @param partitionOnY the partition on Y
 	 * @return true, if successful
 	 * @throws Exception the exception
 	 */
 	public boolean SaveRasterImageAsHadoopFile(JavaPairRDD<Integer,ImageSerializableWrapper> distributedImage, final String outputPath, final ImageType imageType, final int zoomLevel, final int partitionOnX, final int partitionOnY) throws Exception
 	{
 		logger.info("[Babylon][SaveRasterImageAsHadoopFile][Start]");
 		for(int i=0;i<partitionOnX*partitionOnY;i++) {
 			deleteHadoopFile(outputPath+"-"+RasterizationUtils.getImageTileName(zoomLevel,partitionOnX, partitionOnY,i)+".", imageType);
 		}
 		distributedImage.foreach(new VoidFunction<Tuple2<Integer, ImageSerializableWrapper>>() {
 			@Override
 			public void call(Tuple2<Integer, ImageSerializableWrapper> integerImageSerializableWrapperTuple2) throws Exception {
 				SaveRasterImageAsHadoopFile(integerImageSerializableWrapperTuple2._2.image, outputPath+"-"+RasterizationUtils.getImageTileName(zoomLevel,partitionOnX, partitionOnY,integerImageSerializableWrapperTuple2._1), imageType);
 			}
 		});
 		logger.info("[Babylon][SaveRasterImageAsHadoopFile][Stop]");
 		return true;
 	}

 	/**
 	 * Save raster image as S 3 file.
 	 *
 	 * @param distributedImage the distributed image
 	 * @param regionName the region name
 	 * @param accessKey the access key
 	 * @param secretKey the secret key
 	 * @param bucketName the bucket name
 	 * @param path the path
 	 * @param imageType the image type
 	 * @param zoomLevel the zoom level
 	 * @param partitionOnX the partition on X
 	 * @param partitionOnY the partition on Y
 	 * @return true, if successful
 	 */
 	public boolean SaveRasterImageAsS3File(JavaPairRDD<Integer,ImageSerializableWrapper> distributedImage,
 										   final String regionName, final String accessKey, final String secretKey,
 										   final String bucketName, final String path, final ImageType imageType, final int zoomLevel, final int partitionOnX, final int partitionOnY)
 	{
 		logger.info("[Babylon][SaveRasterImageAsS3File][Start]");
 		S3Operator s3Operator = new S3Operator(regionName, accessKey, secretKey);
 		for(int i=0;i<partitionOnX*partitionOnY;i++) {
 			s3Operator.deleteImage(bucketName, path+"-"+RasterizationUtils.getImageTileName(zoomLevel,partitionOnX, partitionOnY,i)+"."+imageType.getTypeName());
 		}
 		distributedImage.foreach(new VoidFunction<Tuple2<Integer, ImageSerializableWrapper>>() {
 			@Override
 			public void call(Tuple2<Integer, ImageSerializableWrapper> integerImageSerializableWrapperTuple2) throws Exception {
 				SaveRasterImageAsS3File(integerImageSerializableWrapperTuple2._2.image, regionName, accessKey, secretKey, bucketName, path+"-"+RasterizationUtils.getImageTileName(zoomLevel,partitionOnX, partitionOnY,integerImageSerializableWrapperTuple2._1), imageType);
 			}
 		});
 		logger.info("[Babylon][SaveRasterImageAsS3File][Stop]");
 		return true;
 	}

 	/**
 	 * Save raster image as local file.
 	 *
 	 * @param rasterImage the raster image
 	 * @param outputPath the output path
 	 * @param imageType the image type
 	 * @return true, if successful
 	 * @throws Exception the exception
 	 */
 	public boolean SaveRasterImageAsLocalFile(BufferedImage rasterImage, String outputPath, ImageType imageType) throws Exception
 	{
 		logger.info("[Babylon][SaveRasterImageAsLocalFile][Start]");
 		File outputImage = new File(outputPath+"."+imageType.getTypeName());
 		outputImage.getParentFile().mkdirs();
 		try {
 			ImageIO.write(rasterImage,imageType.getTypeName(),outputImage);
 		} catch (IOException e) {
 			e.printStackTrace();
 		}
 		logger.info("[Babylon][SaveRasterImageAsLocalFile][Stop]");
 		return true;
 	}

 	/**
 	 * Save raster image as hadoop file.
 	 *
 	 * @param rasterImage the raster image
 	 * @param originalOutputPath the original output path
 	 * @param imageType the image type
 	 * @return true, if successful
 	 * @throws Exception the exception
 	 */
 	public boolean SaveRasterImageAsHadoopFile(BufferedImage rasterImage, String originalOutputPath, ImageType imageType) throws Exception
 	{
 		logger.info("[Babylon][SaveRasterImageAsHadoopFile][Start]");
 		// Locate HDFS path
 		String outputPath = originalOutputPath+"."+imageType.getTypeName();
 		String[] splitString = outputPath.split(":");
 		String hostName = splitString[0]+":"+splitString[1];
 		String[] portAndPath = splitString[2].split("/");
 		String port = portAndPath[0];
 		String localPath = "";
 		for(int i=1;i<portAndPath.length;i++)
 		{
 			localPath+="/"+portAndPath[i];
 		}
 		localPath+="."+imageType.getTypeName();
 		// Delete existing files
 		Configuration hadoopConf = new org.apache.hadoop.conf.Configuration();
 		logger.info("[Babylon][SaveRasterImageAsSparkFile] HDFS URI BASE: "+hostName+":"+port);
 		FileSystem hdfs = FileSystem.get(new URI(hostName+":"+port), hadoopConf);
 		logger.info("[Babylon][SaveRasterImageAsSparkFile] Check the existence of path: "+localPath);
 		if (hdfs.exists(new org.apache.hadoop.fs.Path(localPath))) {
 			logger.info("[Babylon][SaveRasterImageAsSparkFile] Deleting path: "+localPath);
 			hdfs.delete(new org.apache.hadoop.fs.Path(localPath), true);
 			logger.info("[Babylon][SaveRasterImageAsSparkFile] Deleted path: "+localPath);
 		}
 		Path path = new Path(outputPath);
 		FSDataOutputStream out = hdfs.create(path);
 		ImageIO.write(rasterImage,"png",out);
 		out.close();
 		hdfs.close();
 		logger.info("[Babylon][SaveRasterImageAsHadoopFile][Stop]");
 		return true;
 	}


 	/**
 	 * Save raster image as S 3 file.
 	 *
 	 * @param rasterImage the raster image
 	 * @param regionName the region name
 	 * @param accessKey the access key
 	 * @param secretKey the secret key
 	 * @param bucketName the bucket name
 	 * @param path the path
 	 * @param imageType the image type
 	 * @return true, if successful
 	 * @throws IOException Signals that an I/O exception has occurred.
 	 */
 	public boolean SaveRasterImageAsS3File(BufferedImage rasterImage, String regionName, String accessKey, String secretKey, String bucketName, String path, ImageType imageType) throws IOException {
 		logger.info("[Babylon][SaveRasterImageAsS3File][Start]");
 		S3Operator s3Operator = new S3Operator(regionName,accessKey, secretKey);
 		s3Operator.putImage(bucketName,path+"."+imageType.getTypeName(),rasterImage);
 		logger.info("[Babylon][SaveRasterImageAsS3File][Stop]");
 		return true;
 	}


 	/**
 	 * Save vector image as local file.
 	 *
 	 * @param distributedImage the distributed image
 	 * @param outputPath the output path
 	 * @param imageType the image type
 	 * @return true, if successful
 	 * @throws Exception the exception
 	 */
 	public boolean SaveVectorImageAsLocalFile(JavaPairRDD<Integer,String> distributedImage, String outputPath, ImageType imageType) throws Exception
 	{
 		logger.info("[Babylon][SaveVectormageAsLocalFile][Start]");
 		JavaRDD<String> distributedVectorImageNoKey= distributedImage.map(new Function<Tuple2<Integer,String>, String>()
 		{

 			@Override
 			public String call(Tuple2<Integer, String> vectorObject) throws Exception {
 				return vectorObject._2();
 			}

 		});
 		this.SaveVectorImageAsLocalFile(distributedVectorImageNoKey.collect(), outputPath, imageType);
 		logger.info("[Babylon][SaveVectormageAsLocalFile][Stop]");
 		return true;
 	}

 	/**
 	 * Save vector image as local file.
 	 *
 	 * @param vectorImage the vector image
 	 * @param outputPath the output path
 	 * @param imageType the image type
 	 * @return true, if successful
 	 * @throws Exception the exception
 	 */
 	public boolean SaveVectorImageAsLocalFile(List<String> vectorImage, String outputPath, ImageType imageType) throws Exception
 	{
 		logger.info("[Babylon][SaveVectorImageAsLocalFile][Start]");
 		File outputImage = new File(outputPath+"."+imageType.getTypeName());
 		outputImage.getParentFile().mkdirs();

 		BufferedWriter bw = null;
 		FileWriter fw = null;
 		try {
 			fw = new FileWriter(outputImage);
 			bw = new BufferedWriter(fw);
 			for(String svgElement : vectorImage)
 			{
 				bw.write(svgElement);
 			}

 		} catch (IOException e) {

 			e.printStackTrace();

 		} finally {

 			try {

 				if (bw != null)
 					bw.close();

 				if (fw != null)
 					fw.close();

 			} catch (IOException ex) {

 				ex.printStackTrace();

 			}
 		}
 		logger.info("[Babylon][SaveVectorImageAsLocalFile][Stop]");
 		return true;
 	}

 	/**
 	 * Delete hadoop file.
 	 *
 	 * @param originalOutputPath the original output path
 	 * @param imageType the image type
 	 * @return true, if successful
 	 * @throws Exception the exception
 	 */
 	public boolean deleteHadoopFile(String originalOutputPath, ImageType imageType) throws Exception {
 		String outputPath = originalOutputPath+"."+imageType.getTypeName();
 		String[] splitString = outputPath.split(":");
 		String hostName = splitString[0]+":"+splitString[1];
 		String[] portAndPath = splitString[2].split("/");
 		String port = portAndPath[0];
 		String localPath = "";
 		for(int i=1;i<portAndPath.length;i++)
 		{
 			localPath+="/"+portAndPath[i];
 		}
 		localPath+="."+imageType.getTypeName();
 		// Delete existing files
 		Configuration hadoopConf = new org.apache.hadoop.conf.Configuration();
 		logger.info("[Babylon][SaveRasterImageAsSparkFile] HDFS URI BASE: "+hostName+":"+port);
 		FileSystem hdfs = FileSystem.get(new URI(hostName+":"+port), hadoopConf);
 		logger.info("[Babylon][SaveRasterImageAsSparkFile] Check the existence of path: "+localPath);
 		if (hdfs.exists(new org.apache.hadoop.fs.Path(localPath))) {
 			logger.info("[Babylon][SaveRasterImageAsSparkFile] Deleting path: "+localPath);
 			hdfs.delete(new org.apache.hadoop.fs.Path(localPath), true);
 			logger.info("[Babylon][SaveRasterImageAsSparkFile] Deleted path: "+localPath);
 		}
 		return true;
 	}

 	/**
 	 * Delete local file.
 	 *
 	 * @param originalOutputPath the original output path
 	 * @param imageType the image type
 	 * @return true, if successful
 	 */
 	public boolean deleteLocalFile(String originalOutputPath, ImageType imageType) {
 		File file = null;
 		try {

 			// create new file
 			file = new File(originalOutputPath+"."+imageType.getTypeName());

 			// tries to delete a non-existing file
 			file.delete();

 		} catch(Exception e) {

 			// if any error occurs
 			e.printStackTrace();
 		}
 		return true;
 	}
 }
	/**
	* FILE: ImageGenerator.java
	* PATH: org.datasyslab.babylon.core.ImageGenerator.java
	* Copyright (c) 2015-2017 GeoSpark Development Team
	* All rights reserved.
	*/
	package org.datasyslab.babylon.core;

	import java.awt.image.BufferedImage;

	import java.io.*;
	import java.net.URI;
	import java.util.List;

	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.fs.FSDataOutputStream;
	import org.apache.hadoop.fs.FileSystem;
	import org.apache.hadoop.fs.Path;
	import org.apache.log4j.Logger;
	import org.apache.spark.api.java.JavaPairRDD;
	import org.apache.spark.api.java.JavaRDD;
	import org.apache.spark.api.java.function.Function;
	import org.apache.spark.api.java.function.VoidFunction;

	import org.datasyslab.babylon.utils.ImageType;
	import org.datasyslab.babylon.utils.RasterizationUtils;
	import org.datasyslab.babylon.utils.S3Operator;
	import scala.Tuple2;

	import javax.imageio.ImageIO;


	// TODO: Auto-generated Javadoc
	/**
	* The Class ImageGenerator.
	*/
	public class ImageGenerator implements Serializable{

	/** The Constant logger. */
	final static Logger logger = Logger.getLogger(ImageGenerator.class);

	/**
	* Save raster image as local file.
	*
	* @param distributedImage the distributed image
	* @param outputPath the output path
	* @param imageType the image type
	* @param zoomLevel the zoom level
	* @param partitionOnX the partition on X
	* @param partitionOnY the partition on Y
	* @return true, if successful
	* @throws Exception the exception
	*/
	public boolean SaveRasterImageAsLocalFile(JavaPairRDD<Integer,ImageSerializableWrapper> distributedImage, final String outputPath, final ImageType imageType, final int zoomLevel, final int partitionOnX, final int partitionOnY) throws Exception
	{
	logger.info("[Babylon][SaveRasterImageAsLocalFile][Start]");
	for(int i=0;i<partitionOnX*partitionOnY;i++) {
	deleteLocalFile(outputPath+"-"+ RasterizationUtils.getImageTileName(zoomLevel,partitionOnX, partitionOnY,i),imageType);
	}
	distributedImage.foreach(new VoidFunction<Tuple2<Integer, ImageSerializableWrapper>>() {
	@Override
	public void call(Tuple2<Integer, ImageSerializableWrapper> integerImageSerializableWrapperTuple2) throws Exception {
	SaveRasterImageAsLocalFile(integerImageSerializableWrapperTuple2._2.image, outputPath+"-"+RasterizationUtils.getImageTileName(zoomLevel,partitionOnX, partitionOnY,integerImageSerializableWrapperTuple2._1), imageType);
	}
	});
	logger.info("[Babylon][SaveRasterImageAsLocalFile][Stop]");
	return true;
	}

	/**
	* Save raster image as local file.
	*
	* @param distributedImage the distributed image
	* @param outputPath the output path
	* @param imageType the image type
	* @return true, if successful
	* @throws Exception the exception
	*/
	public boolean SaveRasterImageAsLocalFile(JavaPairRDD<Integer,ImageSerializableWrapper> distributedImage, String outputPath, ImageType imageType) throws Exception
	{
	List<Tuple2<Integer,ImageSerializableWrapper>> imagePartitions = distributedImage.collect();
	for(Tuple2<Integer,ImageSerializableWrapper> imagePartition:imagePartitions)
	{
	SaveRasterImageAsLocalFile(imagePartition._2.image, outputPath+"-"+imagePartition._1, imageType);
	}
	return true;
	}

	/**
	* Save raster image as hadoop file.
	*
	* @param distributedImage the distributed image
	* @param outputPath the output path
	* @param imageType the image type
	* @param zoomLevel the zoom level
	* @param partitionOnX the partition on X
	* @param partitionOnY the partition on Y
	* @return true, if successful
	* @throws Exception the exception
	*/
	public boolean SaveRasterImageAsHadoopFile(JavaPairRDD<Integer,ImageSerializableWrapper> distributedImage, final String outputPath, final ImageType imageType, final int zoomLevel, final int partitionOnX, final int partitionOnY) throws Exception
	{
	logger.info("[Babylon][SaveRasterImageAsHadoopFile][Start]");
	for(int i=0;i<partitionOnX*partitionOnY;i++) {
	deleteHadoopFile(outputPath+"-"+RasterizationUtils.getImageTileName(zoomLevel,partitionOnX, partitionOnY,i)+".", imageType);
	}
	distributedImage.foreach(new VoidFunction<Tuple2<Integer, ImageSerializableWrapper>>() {
	@Override
	public void call(Tuple2<Integer, ImageSerializableWrapper> integerImageSerializableWrapperTuple2) throws Exception {
	SaveRasterImageAsHadoopFile(integerImageSerializableWrapperTuple2._2.image, outputPath+"-"+RasterizationUtils.getImageTileName(zoomLevel,partitionOnX, partitionOnY,integerImageSerializableWrapperTuple2._1), imageType);
	}
	});
	logger.info("[Babylon][SaveRasterImageAsHadoopFile][Stop]");
	return true;
	}

	/**
	* Save raster image as S 3 file.
	*
	* @param distributedImage the distributed image
	* @param regionName the region name
	* @param accessKey the access key
	* @param secretKey the secret key
	* @param bucketName the bucket name
	* @param path the path
	* @param imageType the image type
	* @param zoomLevel the zoom level
	* @param partitionOnX the partition on X
	* @param partitionOnY the partition on Y
	* @return true, if successful
	*/
	public boolean SaveRasterImageAsS3File(JavaPairRDD<Integer,ImageSerializableWrapper> distributedImage,
	final String regionName, final String accessKey, final String secretKey,
	final String bucketName, final String path, final ImageType imageType, final int zoomLevel, final int partitionOnX, final int partitionOnY)
	{
	logger.info("[Babylon][SaveRasterImageAsS3File][Start]");
	S3Operator s3Operator = new S3Operator(regionName, accessKey, secretKey);
	for(int i=0;i<partitionOnX*partitionOnY;i++) {
	s3Operator.deleteImage(bucketName, path+"-"+RasterizationUtils.getImageTileName(zoomLevel,partitionOnX, partitionOnY,i)+"."+imageType.getTypeName());
	}
	distributedImage.foreach(new VoidFunction<Tuple2<Integer, ImageSerializableWrapper>>() {
	@Override
	public void call(Tuple2<Integer, ImageSerializableWrapper> integerImageSerializableWrapperTuple2) throws Exception {
	SaveRasterImageAsS3File(integerImageSerializableWrapperTuple2._2.image, regionName, accessKey, secretKey, bucketName, path+"-"+RasterizationUtils.getImageTileName(zoomLevel,partitionOnX, partitionOnY,integerImageSerializableWrapperTuple2._1), imageType);
	}
	});
	logger.info("[Babylon][SaveRasterImageAsS3File][Stop]");
	return true;
	}

	/**
	* Save raster image as local file.
	*
	* @param rasterImage the raster image
	* @param outputPath the output path
	* @param imageType the image type
	* @return true, if successful
	* @throws Exception the exception
	*/
	public boolean SaveRasterImageAsLocalFile(BufferedImage rasterImage, String outputPath, ImageType imageType) throws Exception
	{
	logger.info("[Babylon][SaveRasterImageAsLocalFile][Start]");
	File outputImage = new File(outputPath+"."+imageType.getTypeName());
	outputImage.getParentFile().mkdirs();
	try {
	ImageIO.write(rasterImage,imageType.getTypeName(),outputImage);
	} catch (IOException e) {
	e.printStackTrace();
	}
	logger.info("[Babylon][SaveRasterImageAsLocalFile][Stop]");
	return true;
	}

	/**
	* Save raster image as hadoop file.
	*
	* @param rasterImage the raster image
	* @param originalOutputPath the original output path
	* @param imageType the image type
	* @return true, if successful
	* @throws Exception the exception
	*/
	public boolean SaveRasterImageAsHadoopFile(BufferedImage rasterImage, String originalOutputPath, ImageType imageType) throws Exception
	{
	logger.info("[Babylon][SaveRasterImageAsHadoopFile][Start]");
	// Locate HDFS path
	String outputPath = originalOutputPath+"."+imageType.getTypeName();
	String[] splitString = outputPath.split(":");
	String hostName = splitString[0]+":"+splitString[1];
	String[] portAndPath = splitString[2].split("/");
	String port = portAndPath[0];
	String localPath = "";
	for(int i=1;i<portAndPath.length;i++)
	{
	localPath+="/"+portAndPath[i];
	}
	localPath+="."+imageType.getTypeName();
	// Delete existing files
	Configuration hadoopConf = new org.apache.hadoop.conf.Configuration();
	logger.info("[Babylon][SaveRasterImageAsSparkFile] HDFS URI BASE: "+hostName+":"+port);
	FileSystem hdfs = FileSystem.get(new URI(hostName+":"+port), hadoopConf);
	logger.info("[Babylon][SaveRasterImageAsSparkFile] Check the existence of path: "+localPath);
	if (hdfs.exists(new org.apache.hadoop.fs.Path(localPath))) {
	logger.info("[Babylon][SaveRasterImageAsSparkFile] Deleting path: "+localPath);
	hdfs.delete(new org.apache.hadoop.fs.Path(localPath), true);
	logger.info("[Babylon][SaveRasterImageAsSparkFile] Deleted path: "+localPath);
	}
	Path path = new Path(outputPath);
	FSDataOutputStream out = hdfs.create(path);
	ImageIO.write(rasterImage,"png",out);
	out.close();
	hdfs.close();
	logger.info("[Babylon][SaveRasterImageAsHadoopFile][Stop]");
	return true;
	}



	/**
	* Save raster image as S 3 file.
	*
	* @param rasterImage the raster image
	* @param regionName the region name
	* @param accessKey the access key
	* @param secretKey the secret key
	* @param bucketName the bucket name
	* @param path the path
	* @param imageType the image type
	* @return true, if successful
	* @throws IOException Signals that an I/O exception has occurred.
	*/
	public boolean SaveRasterImageAsS3File(BufferedImage rasterImage, String regionName, String accessKey, String secretKey, String bucketName, String path, ImageType imageType) throws IOException {
	logger.info("[Babylon][SaveRasterImageAsS3File][Start]");
	S3Operator s3Operator = new S3Operator(regionName,accessKey, secretKey);
	s3Operator.putImage(bucketName,path+"."+imageType.getTypeName(),rasterImage);
	logger.info("[Babylon][SaveRasterImageAsS3File][Stop]");
	return true;
	}


	/**
	* Save vector image as local file.
	*
	* @param distributedImage the distributed image
	* @param outputPath the output path
	* @param imageType the image type
	* @return true, if successful
	* @throws Exception the exception
	*/
	public boolean SaveVectorImageAsLocalFile(JavaPairRDD<Integer,String> distributedImage, String outputPath, ImageType imageType) throws Exception
	{
	logger.info("[Babylon][SaveVectormageAsLocalFile][Start]");
	JavaRDD<String> distributedVectorImageNoKey= distributedImage.map(new Function<Tuple2<Integer,String>, String>()
	{

	@Override
	public String call(Tuple2<Integer, String> vectorObject) throws Exception {
	return vectorObject._2();
	}

	});
	this.SaveVectorImageAsLocalFile(distributedVectorImageNoKey.collect(), outputPath, imageType);
	logger.info("[Babylon][SaveVectormageAsLocalFile][Stop]");
	return true;
	}

	/**
	* Save vector image as local file.
	*
	* @param vectorImage the vector image
	* @param outputPath the output path
	* @param imageType the image type
	* @return true, if successful
	* @throws Exception the exception
	*/
	public boolean SaveVectorImageAsLocalFile(List<String> vectorImage, String outputPath, ImageType imageType) throws Exception
	{
	logger.info("[Babylon][SaveVectorImageAsLocalFile][Start]");
	File outputImage = new File(outputPath+"."+imageType.getTypeName());
	outputImage.getParentFile().mkdirs();

	BufferedWriter bw = null;
	FileWriter fw = null;
	try {
	fw = new FileWriter(outputImage);
	bw = new BufferedWriter(fw);
	for(String svgElement : vectorImage)
	{
	bw.write(svgElement);
	}

	} catch (IOException e) {

	e.printStackTrace();

	} finally {

	try {

	if (bw != null)
	bw.close();

	if (fw != null)
	fw.close();

	} catch (IOException ex) {

	ex.printStackTrace();

	}
	}
	logger.info("[Babylon][SaveVectorImageAsLocalFile][Stop]");
	return true;
	}

	/**
	* Delete hadoop file.
	*
	* @param originalOutputPath the original output path
	* @param imageType the image type
	* @return true, if successful
	* @throws Exception the exception
	*/
	public boolean deleteHadoopFile(String originalOutputPath, ImageType imageType) throws Exception {
	String outputPath = originalOutputPath+"."+imageType.getTypeName();
	String[] splitString = outputPath.split(":");
	String hostName = splitString[0]+":"+splitString[1];
	String[] portAndPath = splitString[2].split("/");
	String port = portAndPath[0];
	String localPath = "";
	for(int i=1;i<portAndPath.length;i++)
	{
	localPath+="/"+portAndPath[i];
	}
	localPath+="."+imageType.getTypeName();
	// Delete existing files
	Configuration hadoopConf = new org.apache.hadoop.conf.Configuration();
	logger.info("[Babylon][SaveRasterImageAsSparkFile] HDFS URI BASE: "+hostName+":"+port);
	FileSystem hdfs = FileSystem.get(new URI(hostName+":"+port), hadoopConf);
	logger.info("[Babylon][SaveRasterImageAsSparkFile] Check the existence of path: "+localPath);
	if (hdfs.exists(new org.apache.hadoop.fs.Path(localPath))) {
	logger.info("[Babylon][SaveRasterImageAsSparkFile] Deleting path: "+localPath);
	hdfs.delete(new org.apache.hadoop.fs.Path(localPath), true);
	logger.info("[Babylon][SaveRasterImageAsSparkFile] Deleted path: "+localPath);
	}
	return true;
	}

	/**
	* Delete local file.
	*
	* @param originalOutputPath the original output path
	* @param imageType the image type
	* @return true, if successful
	*/
	public boolean deleteLocalFile(String originalOutputPath, ImageType imageType) {
	File file = null;
	try {

	// create new file
	file = new File(originalOutputPath+"."+imageType.getTypeName());

	// tries to delete a non-existing file
	file.delete();

	} catch(Exception e) {

	// if any error occurs
	e.printStackTrace();
	}
	return true;
	}
	}