How to list files in S3 bucket using Spark Session?

Question

Is it possible to list all of the files in given S3 path (ex: s3://my-bucket/my-folder/*.extension) using a SparkSession object?

Answer 1

You can use Hadoop API for accessing files on S3 (Spark uses it as well):

import java.net.URI
import org.apache.hadoop.fs.FileSystem
import org.apache.hadoop.fs.Path
import org.apache.hadoop.conf.Configuration

val path = "s3://somebucket/somefolder"
val fileSystem = FileSystem.get(URI.create(path), new Configuration())
val it = fileSystem.listFiles(new Path(path), true)
while (it.hasNext()) {
  ...
}

Answer 2

Approach 1

For pyspark users, I've translated Michael Spector's answer (I'll leave it to you to decide if using this is a good idea):

sc = spark.sparkContext
myPath = f's3://my-bucket/my-prefix/'
javaPath = sc._jvm.java.net.URI.create(myPath)
hadoopPath = sc._jvm.org.apache.hadoop.fs.Path(myPath)
hadoopFileSystem = sc._jvm.org.apache.hadoop.fs.FileSystem.get(javaPath, sc._jvm.org.apache.hadoop.conf.Configuration())
iterator = hadoopFileSystem.listFiles(hadoopPath, True)

s3_keys = []
while iterator.hasNext():
    s3_keys.append(iterator.next().getPath().toUri().getRawPath())    

s3_keys now holds all file keys found at my-bucket/my-prefix

Approach 2 Here is an alternative that I found (hat tip to @forgetso):

myPath = 's3://my-bucket/my-prefix/*'
hadoopPath = sc._jvm.org.apache.hadoop.fs.Path(myPath)
hadoopFs = hadoopPath.getFileSystem(sc._jvm.org.apache.hadoop.conf.Configuration())
statuses = hadoopFs.globStatus(hadoopPath)

for status in statuses:
  status.getPath().toUri().getRawPath()
  # Alternatively, you can get file names only with:
  # status.getPath().getName()

Approach 3 (incomplete!)

The two approaches above do not use the Spark parallelism mechanism that would be applied on a distributed read. That logic looks private though. See parallelListLeafFiles here. I have not found a way to compel pyspark do to a distributed ls on s3 without also reading the file contents. I tried to use py4j to instantiate an InMemoryFileIndex, but can't get the incantation right. Here is what I have so far if someone wants to pick it up from here:

myPath = f's3://my-bucket/my-path/'
paths = sc._gateway.new_array(sc._jvm.org.apache.hadoop.fs.Path, 1)
paths[0] = sc._jvm.org.apache.hadoop.fs.Path(myPath)

emptyHashMap = sc._jvm.java.util.HashMap()
emptyScalaMap = sc._jvm.scala.collection.JavaConversions.mapAsScalaMap(emptyMap)

# Py4J is not happy with this:
sc._jvm.org.apache.spark.sql.execution.datasources.InMemoryFileIndex(
    spark._jsparkSession, 
    paths, 
    emptyScalaMap, 
    sc._jvm.scala.Option.empty() # Optional None
)

Answer 3

Lou Zell was very close! The below ended up working on ADLS2, but I'm putting it here because of the Py4J magic. Note that the noopcache causes the job to be run twice: Once when the index is created and once when listfiles is called. I also wrote a blogpost on this: https://www.perceptivebits.com/a-comprehensive-guide-to-finding-files-via-spark/

import os

base_path = "/mnt/my_data/"
glob_pattern = "*"
sc = spark.sparkContext
hadoop_base_path = sc._jvm.org.apache.hadoop.fs.Path(base_path)
paths = sc._jvm.PythonUtils.toSeq([hadoop_base_path])

#noop_cache_clazz = sc._jvm.java.lang.Class.forName("org.apache.spark.sql.execution.datasources.NoopCache$")
#ff = noop_cache_clazz.getDeclaredField("MODULE$")
#noop_cache = ff.get(None)

file_status_cache_clazz = jvm.java.lang.Class.forName(
            "org.apache.spark.sql.execution.datasources.FileStatusCache$"
        )
ff = file_status_cache_clazz.getDeclaredField("MODULE$")
jvm_spark_session = spark._jsparkSession
file_status_cache = ff.get(None).getOrCreate(jvm_spark_session)

in_memory_file_index = sc._jvm.org.apache.spark.sql.execution.datasources.InMemoryFileIndex(
    jvm_spark_session,
    paths,
    sc._jvm.PythonUtils.toScalaMap({}),
    sc._jvm.scala.Option.empty(),
    file_status_catche,  # or use noop_cache if you need to save memory
    sc._jvm.scala.Option.empty(),
    sc._jvm.scala.Option.empty()
)
glob_path = sc._jvm.org.apache.hadoop.fs.Path(os.path.join(base_path, glob_pattern))
glob_paths = sc._jvm.PythonUtils.toSeq([glob_path])
# SparkHadoopUtil.get.globPath(fs, Path.mergePaths(validated(basep), validated(globp))),
status_list = in_memory_file_index.listLeafFiles(glob_paths)
path_list = []
iter = status_list.iterator()
while iter.hasNext():
    path_status = iter.next()
    path_list.append(str(path_status.getPath().toUri().getRawPath()))

path_list.sort()

print(path_list)

Answer 4

You can use input_file_name with dataframe, it will give you absolute file-path per row.

Following code will give you all the file paths.

spark.read.table("zen.intent_master").select(input_file_name).distinct.collect

I am assuming. For your use case, you just want to read data from a set of files, with some regex, so then you can apply that in filter.

For example,

val df = spark.read.table("zen.intent_master").filter(input_file_name.rlike("your regex string"))