package test
import java.util.HashMap
import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
import org.apache.spark.streaming._
import org.apache.spark.sql.SparkSession
import com.ibm.ibmos2spark.CloudObjectStorage
import org.apache.spark.sql.execution.datasources.csv.CSVFileFormat
import com.ibm.stocator.fs.ObjectStoreFileSystem

object SparkCosFinalSL {

  def main(args: Array[String]) {

    val conf = new SparkConf().setAppName("test").setMaster("local[*]").set("spark.driver.host", "localhost")

    val sc = new SparkContext(conf)

   /*val credentials = scala.collection.mutable.HashMap[String, String](
      "accessKey" -> "L_-uMLV9AU-ZBWr0BE6JmiHMYFqsORXndMmfrpaqJIgG", 
      "secretKey" -> "crn:v1:bluemix:public:cloud-object-storage:global:a/647b189897a37a7ac4dbf0a3ef43fc42:866ec777-5c98-4e1c-b2bf-e5d0b1d13694::",
      //"crn:v1:bluemix:public:iam-identity::a/647b189897a37a7ac4dbf0a3ef43fc42::serviceid:ServiceId-7fd06371-f1ed-482f-9e12-090a82f713ac", 
      "iamServiceEndpoint" -> "https://iam.bluemix.net/oidc/token",  
      "apiKey" -> "L_-uMLV9AU-ZBWr0BE6JmiHMYFqsORXndMmfrpaqJIgG",
      "serviceId" -> "crn:v1:bluemix:public:cloud-object-storage:global:a/647b189897a37a7ac4dbf0a3ef43fc42:866ec777-5c98-4e1c-b2bf-e5d0b1d13694::",
      "endPoint" -> "https://s3-api.us-geo.objectstorage.softlayer.net"
      
      
    )*/
    
 /*   val service_credentials_COS_from_bluemix = scala.collection.mutable.HashMap[String, String](
    "endPoint" -> "s3-api.us-geo.objectstorage.service.networklayer.com",
    "serviceId" ->"crn:v1:bluemix:public:iam-identity::a/647b189897a37a7ac4dbf0a3ef43fc42::serviceid:ServiceId-7fd06371-f1ed-482f-9e12-090a82f713ac",
    "apiKey"-> "L_-uMLV9AU-ZBWr0BE6JmiHMYFqsORXndMmfrpaqJIgG"
    
    )*/
    
  var credentials = scala.collection.mutable.HashMap[String, String](
    "endPoint"->"s3-api.us-geo.objectstorage.service.networklayer.com",
    "apiKey"->"L_-uMLV9AU-ZBWr0BE6JmiHMYFqsORXndMmfrpaqJIgG",
    "serviceId"->"crn:v1:bluemix:public:iam-identity::a/647b189897a37a7ac4dbf0a3ef43fc42::serviceid:ServiceId-7fd06371-f1ed-482f-9e12-090a82f713ac",
    "accessKey"->"L_-uMLV9AU-ZBWr0BE6JmiHMYFqsORXndMmfrpaqJIgG",
    "secretKey"->"crn:v1:bluemix:public:iam-identity::a/647b189897a37a7ac4dbf0a3ef43fc42::serviceid:ServiceId-7fd06371-f1ed-482f-9e12-090a82f713ac",
    "endPoint"->"http://s3-api.us-geo.objectstorage.softlayer.net",
    "iamServiceEndpoint"->"https://iam.ng.bluemix.net/oidc/token",
    "region"->"us-geo"
)


    var bucketName = "tests"
    var objectname = "temperatureUS.csv"
    
     var mConf = sc.hadoopConfiguration
   
    // global Stocator-COS definitions
    mConf.set("fs.stocator.scheme.list", "cos");
    mConf.set("fs.cos.impl", "com.ibm.stocator.fs.ObjectStoreFileSystem");
    mConf.set("fs.stocator.cos.impl", "com.ibm.stocator.fs.cos.COSAPIClient");
    mConf.set("fs.stocator.cos.scheme", "cos");

    // "us-geo" is the service name. Can be any other name
    // All configuration keys will have the prefix fs.cos.us-geo


 //Usage with access key and secret key
    mConf.set("fs.cos.myCos.endpoint", "http://s3-api.us-geo.objectstorage.softlayer.net");
    mConf.set("fs.cos.myCos.v2.signer.type", "false")  
    mConf.set("fs.cos.myCos.access.key", "L_-uMLV9AU-ZBWr0BE6JmiHMYFqsORXndMmfrpaqJIgG");
    mConf.set("fs.cos.myCos.secret.key", "crn:v1:bluemix:public:iam-identity::a/647b189897a37a7ac4dbf0a3ef43fc42::serviceid:ServiceId-7fd06371-f1ed-482f-9e12-090a82f713ac");
   
    var configurationName = "myCos" // you can choose any string you want
    var cos = new CloudObjectStorage(sc, credentials,
                                configurationName=configurationName,
                                cosType="bluemix_cos") // cosType = "bluemix_cos"

    var spark = SparkSession.
      builder().appName("test").
        getOrCreate()
    spark.sparkContext.setLogLevel("TRACE")

    var dfData1 = spark.
      read.format("org.apache.spark.sql.execution.datasources.csv.CSVFileFormat").
      option("header", "true").
      option("inferSchema", "true").
     // textFile(cos.url(bucketName, objectname))
    load(cos.url(bucketName, objectname))

    dfData1.foreach(println(_))

    

  }
}