title: Spark dataframe databricks csv appends extra double quotes
body: <p>It seems that when I apply <code>CONCAT</code> on a <code>dataframe</code> in <code>spark sql</code> and store that <code>dataframe</code> as csv file in a <code>HDFS</code> location, then  there are extra double quotes added to that <code>concat</code> column alone in the ouput file .</p>
  
  <p>This double quotes are not added when I appy show.This double quotes are added only when I store that <code>dataframe</code> as a csv file </p>
  
  <p>It seems that I need to remove the  extra double quotes that gets added when I save the <code>dataframe</code> as csv file.</p>
  
  <p>I am using the <code>com.databricks:spark-csv_2.10:1.1.0</code> jar </p>
  
  <p>Spark Version is 1.5.0-cdh5.5.1</p>
  
  <p><strong>Input :</strong></p>
  
  <pre><code> campaign_file_name_1, campaign_name_1, shagdhsjagdhjsagdhrSqpaKa5saoaus89,    1
   campaign_file_name_1, campaign_name_1, sagdhsagdhasjkjkasihdklas872hjsdjk,    2
  </code></pre>
  
  <p>Expected output :</p>
  
  <pre><code> campaign_file_name_1, shagdhsjagdhjsagdhrSqpaKa5saoaus89,     campaign_name_1"="1,  2017-06-06 17:09:31
   campaign_file_name_1, sagdhsagdhasjkjkasihdklas872hjsdjk,   campaign_name_1"="2,  2017-06-06 17:09:31
  </code></pre>
  
  <p><strong>Spark Code :</strong></p>
  
  <pre><code>  object campaignResultsMergerETL extends BaseETL {
  
    val now  = ApplicationUtil.getCurrentTimeStamp()
    val conf = new Configuration()
    val fs  = FileSystem.get(conf)
    val log = LoggerFactory.getLogger(this.getClass.getName)
  
    def main(args: Array[String]): Unit = {
      //---------------------
      code for sqlContext Initialization 
      //---------------------
      val campaignResultsDF  = sqlContext.read.format("com.databricks.spark.avro").load(campaignResultsLoc)
      campaignResultsDF.registerTempTable("campaign_results")
      val campaignGroupedDF =  sqlContext.sql(
     """
      |SELECT campaign_file_name,
      |campaign_name,
      |tracker_id,
      |SUM(campaign_measure) AS campaign_measure
      |FROM campaign_results
      |GROUP BY campaign_file_name,campaign_name,tracker_id
    """.stripMargin)
  
      campaignGroupedDF.registerTempTable("campaign_results_full")
  
      val campaignMergedDF =  sqlContext.sql(
    s"""
      |SELECT campaign_file_name,
      |tracker_id,
      |CONCAT(campaign_name,'\"=\"' ,campaign_measure),
      |"$now" AS audit_timestamp
      |FROM campaign_results_full
    """.stripMargin)
  
     campaignMergedDF.show(20)
     saveAsCSVFiles(campaignMergedDF, campaignResultsExportLoc, numPartitions)
  
     }
  
  
      def saveAsCSVFiles(campaignMeasureDF:DataFrame,hdfs_output_loc:String,numPartitions:Int): Unit =
      {
         log.info("saveAsCSVFile method started")
         if (fs.exists(new Path(hdfs_output_loc))){
            fs.delete(new Path(hdfs_output_loc), true)
         }
       campaignMeasureDF.repartition(numPartitions).write.format("com.databricks.spark.csv").save(hdfs_output_loc)
         log.info("saveAsCSVFile method ended")
      }
  
   }
  </code></pre>
  
  <p>Result of <code>campaignMergedDF.show(20)</code> is correct and works fine.</p>
  
  <pre><code> campaign_file_name_1, shagdhsjagdhjsagdhrSqpaKa5saoaus89,   campaign_name_1"="1,  2017-06-06 17:09:31
   campaign_file_name_1, sagdhsagdhasjkjkasihdklas872hjsdjk,   campaign_name_1"="2,  2017-06-06 17:09:31
  </code></pre>
  
  <p>Result of <code>saveAsCSVFiles</code> : This is incorrect .</p>
  
  <pre><code> campaign_file_name_1, shagdhsjagdhjsagdhrSqpaKa5saoaus89,   "campaign_name_1""=""1",  2017-06-06 17:09:31
   campaign_file_name_1, sagdhsagdhasjkjkasihdklas872hjsdjk,   "campaign_name_1""=""2",  2017-06-06 17:09:31
  </code></pre>
  
  <p>Could some one help me on this issue?</p>