Rename and Move S3 files based on their folders name in spark scala

Question

I have spark output in a s3 folders and I want to move all s3 files from that output folder to another location ,but while moving I want to rename the files .

For example I have files in S3 folders like below

Now I want to rename all files and put into another directory,but the name of the files would be like below

Fundamental.FinancialStatement.FinancialStatementLineItems.Japan.1971-BAL.1.2017-10-18-0439.Full.txt
Fundamental.FinancialStatement.FinancialStatementLineItems.Japan.1971-BAL.2.2017-10-18-0439.Full.txt
Fundamental.FinancialStatement.FinancialStatementLineItems.Japan.1971-BAL.3.2017-10-18-0439.Full.txt

Here Fundamental.FinancialStatementis constant in all the files 2017-10-18-0439 current date time .

This is what I have tried so far but not able to get folder name and loop through all files

    import org.apache.hadoop.fs._

val src = new Path("s3://trfsmallfffile/Segments/output")
val dest = new Path("s3://trfsmallfffile/Segments/Finaloutput")
val conf = sc.hadoopConfiguration   // assuming sc = spark context
val fs = src.getFileSystem(conf)
//val file = fs.globStatus(new Path("src/DataPartition=Japan/part*.gz"))(0).getPath.getName
//println(file)
val status = fs.listStatus(src)    

status.foreach(filename => {
               val a = filename.getPath.getName.toString()
                println("file name"+a)
                //println(filename)
             })

This gives me below output

    file nameDataPartition=Japan
file nameDataPartition=SelfSourcedPrivate
file nameDataPartition=SelfSourcedPublic
file name_SUCCESS

This gives me folders details not files inside the folder.

Reference is taken from here Stack Overflow Refrence

did you try using `fs.listFiles` and then `map` over it to rename each file? — philantrovert, Jan 16 '18 at 12:45
@philantrovert Yes but extracting folders name and then moving I am not able to do ..Can you please put some sample code — Atharv Thakur, Jan 16 '18 at 12:50
I don't understand. Once you do `fs.rename(src, dest)`, wouldn't `dest` become the folder name under which all the files are that need to be renamed. Apologies if I'm missing something. I'm behind a proxy and I can't see the image. — philantrovert, Jan 16 '18 at 12:52
Arthav: debugging individual lines of code on stack overflow is not a viable development practice. I'd recommend an IDE like IntelliJ IDEA, scalatest for the tests, and setting breakpoints. Your code is broken, which is the kind of thing step-by-step debugging is going to tell you. — stevel, Jan 16 '18 at 14:11
@SteveLoughran yes you are right ..I need to use that ...Currently I am using as script in zeppelin note book .. — Atharv Thakur, Jan 16 '18 at 14:16
Hello Atharv, did this work for you.? please release the solution here so that I can refer it for my issue. thanks — TheCodeCache, Sep 23 '18 at 08:15
was that working in your local or is it only working in aws cloud systems.? — TheCodeCache, Sep 24 '18 at 19:10

score 5 · Accepted Answer · edited Jun 22 '22 at 21:14

You are getting directory because you have sub dir level in s3 .

/*/* to go in subdir .

Try this

import org.apache.hadoop.fs._

val src = new Path("s3://trfsmallfffile/Segments/Output/*/*")
val dest = new Path("s3://trfsmallfffile/Segments/FinalOutput")
val conf = sc.hadoopConfiguration   // assuming sc = spark context
val fs = src.getFileSystem(conf)

val file = fs.globStatus(new Path("s3://trfsmallfffile/Segments/Output/*/*"))


  for (urlStatus <- file) {
    //println("S3 FILE PATH IS ===:" + urlStatus.getPath)
    val partitioName=urlStatus.getPath.toString.split("=")(1).split("\\/")(0).toString
    val finalPrefix="Fundamental.FinancialLineItem.Segments."
    val finalFileName=finalPrefix+partitioName+".txt"
    val dest = new Path("s3://trfsmallfffile/Segments/FinalOutput"+"/"+finalFileName+ " ")
    fs.rename(urlStatus.getPath, dest)
  }

score 0 · Answer 2 · answered Jan 16 '18 at 14:27

0

This has worked for me in past

import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.hadoop.conf.Configuration 
val path = "s3://<bucket>/<directory>"
val fs = FileSystem.get(new java.net.URI(path), spark.sparkContext.hadoopConfiguration)
fs.listStatus(new Path(path))

The list status provides all the files in the s3 directory

answered Jan 16 '18 at 14:27

Avishek Bhattacharya

6,534
3
34
53

no it does not gives me files it lists only folders details – Atharv Thakur Jan 16 '18 at 14:58

Rename and Move S3 files based on their folders name in spark scala

2 Answers2

Linked