Null column and special characters present in col val handled

Author: samuel-pt

src/main/scala/com/springml/spark/salesforce/DataWriter.scala

@@ -62,7 +62,7 @@ class DataWriter (val userName: String, val password: String,
   }
 
   def writeData(rdd: RDD[Row], metadataId: String): Boolean = {
-    val csvRDD = rdd.map(row => row.toSeq.map(value => value.toString).mkString(","))
+    val csvRDD = rdd.map(row => row.toSeq.map(value => Utils.rowValue(value)).mkString(","))
     csvRDD.mapPartitionsWithIndex {
       case (index, iterator) => {
         @transient val logger = Logger.getLogger(classOf[DataWriter])

src/main/scala/com/springml/spark/salesforce/Utils.scala

@@ -91,15 +91,29 @@ object Utils extends Serializable {
   def rowSize(rows: Array[Row]) : Long = {
       var sizeOfRows = 0l
       for (row <- rows) {
-      //for (i <- 0 until rows.length) {
-        val rowSize = SizeEstimator.estimate(row.toSeq.map { value => value.toString() }.mkString(","))
         // Converting to bytes
+        val rowSize = SizeEstimator.estimate(row.toSeq.map { value => rowValue(value) }.mkString(","))
         sizeOfRows += rowSize
       }
 
       sizeOfRows
   }
 
+  def rowValue(rowVal: Any) : String = {
+    if (rowVal == null) {
+      ""
+    } else {
+      var value = rowVal.toString()
+      if (value.contains("\"")) {
+        value = value.replaceAll("\"", "\"\"")
+      }
+      if (value.contains("\"") || value.contains("\n") || value.contains(",")) {
+        value = "\"" + value + "\""
+      }
+      value
+    }
+  }
+
   def metadataConfig(usersMetadataConfig: Option[String]) = {
     var systemMetadataConfig = readMetadataConfig();
     if (usersMetadataConfig != null && usersMetadataConfig.isDefined) {