title: How to rename spark data frame output file in AWS in spark SCALA
body: <p>I am saving my spark data frame output as csv file in scala with partitions.
  This is how i do that in <strong>Zeppelin</strong>.</p>
  
  <pre><code>val sqlContext = new org.apache.spark.sql.SQLContext(sc)
  
      import sqlContext.implicits._
      import org.apache.spark.{ SparkConf, SparkContext }
      import java.sql.{Date, Timestamp}
      import org.apache.spark.sql.Row
      import org.apache.spark.sql.types._
      import org.apache.spark.sql.functions.udf
  
  import org.apache.spark.sql.functions.input_file_name
  import org.apache.spark.sql.functions.regexp_extract
  
  val get_cus_val = spark.udf.register("get_cus_val", (filePath: String) =&gt; filePath.split("\\.")(3))
  
  val rdd = sc.textFile("s3://trfsmallfffile/FinancialLineItem/MAIN")
  val header = rdd.filter(_.contains("LineItem.organizationId")).map(line =&gt; line.split("\\|\\^\\|")).first()
  val schema = StructType(header.map(cols =&gt; StructField(cols.replace(".", "_"), StringType)).toSeq)
  val data = sqlContext.createDataFrame(rdd.filter(!_.contains("LineItem.organizationId")).map(line =&gt; Row.fromSeq(line.split("\\|\\^\\|").toSeq)), schema)
  
  val schemaHeader = StructType(header.map(cols =&gt; StructField(cols.replace(".", "."), StringType)).toSeq)
  val dataHeader = sqlContext.createDataFrame(rdd.filter(!_.contains("LineItem.organizationId")).map(line =&gt; Row.fromSeq(line.split("\\|\\^\\|").toSeq)), schemaHeader)
  
  val df1resultFinal=data.withColumn("DataPartition", get_cus_val(input_file_name))
  val rdd1 = sc.textFile("s3://trfsmallfffile/FinancialLineItem/INCR")
  val header1 = rdd1.filter(_.contains("LineItem.organizationId")).map(line =&gt; line.split("\\|\\^\\|")).first()
  val schema1 = StructType(header1.map(cols =&gt; StructField(cols.replace(".", "_"), StringType)).toSeq)
  val data1 = sqlContext.createDataFrame(rdd1.filter(!_.contains("LineItem.organizationId")).map(line =&gt; Row.fromSeq(line.split("\\|\\^\\|").toSeq)), schema1)
  
  
  import org.apache.spark.sql.expressions._
  val windowSpec = Window.partitionBy("LineItem_organizationId", "LineItem_lineItemId").orderBy($"TimeStamp".cast(LongType).desc) 
  val latestForEachKey = data1.withColumn("rank", rank().over(windowSpec)).filter($"rank" === 1).drop("rank", "TimeStamp")
  
  
  val dfMainOutput = df1resultFinal.join(latestForEachKey, Seq("LineItem_organizationId", "LineItem_lineItemId"), "outer")
        .select($"LineItem_organizationId", $"LineItem_lineItemId",
          when($"DataPartition_1".isNotNull, $"DataPartition_1").otherwise($"DataPartition").as("DataPartition"),
          when($"StatementTypeCode_1".isNotNull, $"StatementTypeCode_1").otherwise($"StatementTypeCode").as("StatementTypeCode"),
          when($"FinancialConceptLocalId_1".isNotNull, $"FinancialConceptLocalId_1").otherwise($"FinancialConceptLocalId").as("FinancialConceptLocalId"),
          when($"FinancialConceptGlobalId_1".isNotNull, $"FinancialConceptGlobalId_1").otherwise($"FinancialConceptGlobalId").as("FinancialConceptGlobalId"),
          when($"FinancialConceptCodeGlobalSecondaryId_1".isNotNull, $"FinancialConceptCodeGlobalSecondaryId_1").otherwise($"FinancialConceptCodeGlobalSecondaryId").as("FinancialConceptCodeGlobalSecondaryId"),
          when($"FFAction_1".isNotNull, $"FFAction_1").otherwise($"FFAction|!|").as("FFAction|!|"))
          .filter(!$"FFAction|!|".contains("D|!|"))
  
  val dfMainOutputFinal = dfMainOutput.na.fill("").select($"DataPartition",$"StatementTypeCode",concat_ws("|^|", dfMainOutput.schema.fieldNames.filter(_ != "DataPartition").map(c =&gt; col(c)): _*).as("concatenated"))
  
  val headerColumn = dataHeader.columns.toSeq
  
  val header = headerColumn.mkString("", "|^|", "|!|").dropRight(3)
  
  val dfMainOutputFinalWithoutNull = dfMainOutputFinal.withColumn("concatenated", regexp_replace(col("concatenated"), "|^|null", "")).withColumnRenamed("concatenated", header)
  
  
  dfMainOutputFinalWithoutNull.repartition(1).write.partitionBy("DataPartition","StatementTypeCode")
    .format("csv")
    .option("nullValue", "")
    .option("delimiter", "\t")
    .option("quote", "\u0000")
    .option("header", "true")
    .option("codec", "gzip")
    .save("s3://trfsmallfffile/FinancialLineItem/output")
  
    val FFRowCount =dfMainOutputFinalWithoutNull.groupBy("DataPartition","StatementTypeCode").count
  
    FFRowCount.coalesce(1).write.format("com.databricks.spark.xml")
    .option("rootTag", "FFFileType")
    .option("rowTag", "FFPhysicalFile")
    .save("s3://trfsmallfffile/FinancialLineItem/Descr")
  </code></pre>
  
  <p>Now files are saved in partitioned folder structure which is expected .</p>
  
  <p>Now my requiremen is to rename all the part file and save it in one directory .
  The name of the file will be as the name of the folder structure .</p>
  
  <p>For example i have one file saved in <code>folder/DataPartition=Japan/PartitionYear=1971/part-00001-87a61115-92c9-4926-a803-b46315e55a08.c000.csv.gz</code></p>
  
  <p>Now i want my file name to be </p>
  
  <pre><code>Japan.1971.1.txt.gz
  Japan.1971.2.txt.gz
  </code></pre>
  
  <p>I have done this in java map-reduce after my job is completed then i was reading HDFS files system and then moved it into different location as renamed file name .</p>
  
  <p>But how do to the this in AWS S3 files system in spark SCALA .</p>
  
  <p>As far as i have research there is no direct way to rename spark data frame output file name.</p>
  
  <p>But there is implementation that can be done  in the job itself using <code>MultipleOutputs</code> as <strong>saveAsHadoopFile</strong> but how to do that ?. </p>
  
  <p>I am looking for some sample code in scala </p>
  
  <p>It is as like after completing job we need to read the file from s3,reame it  and move it to some other location .</p>