Add CloudObjectStorage R class for COS support

Author: bassel-zeidan

r/sparkr/R/osconfig.R

@@ -9,24 +9,24 @@ swifturl = function(name, container_name, object_name){
 
 
 #' sparkcontext is a SparkContext object.
-#' 
+#'
 #' name is a string that identifies this configuration. You can
 #'     use any string you like. This allows you to create
 #'     multiple configurations to different Object Storage accounts.
 #' auth_url, username and password are string credentials for your
 #' Softlayer Object Store
-#' @export softlayer     
+#' @export softlayer
 #' @exportClass softlayer
 
 softlayer <- setRefClass("softlayer",
   fields=list(name="character", container_name="character", object_name="character",
-             sparkcontext='jobj', auth_url="character", 
+             sparkcontext='jobj', auth_url="character",
               tenant = "character", username="character", password="character"),
-  methods=list(initialize = 
+  methods=list(initialize =
     function( sparkcontext, name, auth_url, tenant, username, password,public=FALSE,
-              swift2d_driver='com.ibm.stocator.fs.ObjectStoreFileSystem'){     
+              swift2d_driver='com.ibm.stocator.fs.ObjectStoreFileSystem'){
 
-        .self$name = name 
+        .self$name = name
         prefix = paste("fs.swift2d.service" , name, sep =".")
         hConf = SparkR:::callJMethod(sparkcontext, "hadoopConfiguration")
         SparkR:::callJMethod(hConf, "set", "fs.swift2d.impl", swift2d_driver)
@@ -41,21 +41,21 @@ softlayer <- setRefClass("softlayer",
         SparkR:::callJMethod(hConf, "set", paste(prefix, "use.get.auth", sep='.'), "true")
         invisible(SparkR:::callJMethod(hConf, "setBoolean", paste(prefix, "location-aware", sep='.'), FALSE))
         SparkR:::callJMethod(hConf, "set", paste(prefix, "password", sep='.'), password)
-        
-        
+
+
     },
-               
+
         url = function(container_name, object_name){
         return(swifturl(name, container_name, object_name))}
     )
 )
 
 
-          
+
 #' sparkcontext:  a SparkContext object.
 #'
 #' credentials:  a dictionary with the following required keys:
-#'   
+#'
 #'   auth_url
 #'   project_id (or projectId)
 #'   user_id (or userId)
@@ -73,28 +73,28 @@ softlayer <- setRefClass("softlayer",
 #' instances, the values for these credentials can be obtained
 #' by clicking on the 'insert to code' link just below a data
 #' source.
-#' @export bluemix     
+#' @export bluemix
 #' @exportClass bluemix
 
-          
+
 bluemix <- setRefClass("bluemix",
-  fields=list(name="character", credentials = "list", 
+  fields=list(name="character", credentials = "list",
              sparkcontext='jobj', public = "character"),
-  methods=list(initialize = 
+  methods=list(initialize =
     function(..., sparkcontext, name=NULL, credentials,
-             public=FALSE,swift2d_driver='com.ibm.stocator.fs.ObjectStoreFileSystem'){       
+             public=FALSE,swift2d_driver='com.ibm.stocator.fs.ObjectStoreFileSystem'){
 
       callSuper(...,credentials=credentials)
-        
+
       if ( is.null(name)) name <<- credentials["name"][[1]]
-           
+
       user_id = try( credentials['user_id'][[1]])
       if(class(user_id)=="try-error")  user_id = credentials['userId'][[1]]
-          
+
       tenant = try( credentials['project_id'][[1]])
       if(class(tenant)=="try-error")  tenant = credentials['projectId'][[1]]
-        
-        .self$name = name  
+
+        .self$name = name
         prefix = paste("fs.swift2d.service" , name, sep =".")
         hConf = SparkR:::callJMethod(sparkcontext, "hadoopConfiguration")
         SparkR:::callJMethod(hConf, "set", "fs.swift2d.impl", swift2d_driver)
@@ -108,8 +108,63 @@ bluemix <- setRefClass("bluemix",
         invisible(SparkR:::callJMethod(hConf, "setBoolean", paste(prefix, "public", sep='.'), public))
         #invisible(SparkR:::callJMethod(hConf, "setInt", paste(prefix, "http.port", sep='.'), 8080))
           },
-          
+
         url = function( container_name, object_name){
         return(swifturl(name, container_name, object_name))}
     )
-)
+)
+
+#' CloudObjectStorage is a class that is designed for IBM cloud object storage (COS)
+#' It sets up the hadoop config for COS and provide the final file url.
+#'
+#' sparkContext:  a SparkContext object.
+#''
+#' credentials:  a dictionary with the following required keys:
+#'   endpoint
+#'   accessKey
+#'   secretKey
+#'
+#' configurationName: string identifies the configurations that has been
+#' set.
+#' When using this from a IBM Spark service instance that
+#' is configured to connect to particular Bluemix object store
+#' instances, the values for these credentials can be obtained
+#' by clicking on the 'insert to code' link just below a data
+#' source.
+#' @export bluemix
+#' @exportClass bluemix
+CloudObjectStorage <- setRefClass("CloudObjectStorage",
+  fields=list(configName="character"),
+  methods=list(
+      initialize = function(..., sparkContext, credentials, configurationName){
+
+
+          if (is.null(credentials["endpoint"][[1]])) {
+              stop("Attribute endpoint in credentials is missing!")
+          }
+
+          if (is.null(credentials["accessKey"][[1]])) {
+              stop("Attribute accessKey in credentials is missing!")
+          }
+
+          if (is.null(credentials["secretKey"][[1]])) {
+              stop("Attribute secretKey in credentials is missing!")
+          }
+
+          .self$configName = configurationName
+          prefix = "fs.s3d.service"
+          hConf = SparkR:::callJMethod(sparkContext, "hadoopConfiguration")
+          SparkR:::callJMethod(hConf, "set", paste(prefix, "endpoint", sep='.'), credentials['endpoint'][[1]])
+          SparkR:::callJMethod(hConf, "set", paste(prefix, "access.key", sep='.'), credentials['accessKey'][[1]])
+          SparkR:::callJMethod(hConf, "set", paste(prefix, "secret.key", sep='.'), credentials['secretKey'][[1]])
+      },
+
+      getConfigName = function() {
+        return (.self$configName)
+      },
+
+      url = function(bucketName, objectName){
+          return(paste("s3d://", bucketName, ".service/", objectName, sep = ""))
+      }
+  )
+)