是否可以使用 SparkSession 对象列出给定 S3 路径（例如：s3://my-bucket/my-folder/*.extension）中的所有文件？
【问题讨论】：
标签： apache-spark amazon-s3 apache-spark-sql
您可以将input_file_name 与数据框一起使用，它会为您提供每行的绝对文件路径。
以下代码将为您提供所有文件路径。
spark.read.table("zen.intent_master").select(input_file_name).distinct.collect
我假设。对于您的用例，您只想使用一些正则表达式从一组文件中读取数据，然后您可以在过滤器中应用它。
例如，
val df = spark.read.table("zen.intent_master").filter(input_file_name.rlike("your regex string"))
【讨论】：
val blah = spark.read.json("s3://blah-bucket/blah-folder/") val blahWithFilename = blah.withColumn("filename", input_file_name) blahWithFilename.show(false)
您可以使用 Hadoop API 访问 S3 上的文件（Spark 也使用它）：
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()) {
...
}
【讨论】：
it 是 org.apache.hadoop.fs.RemoteIterator 的一个实例，而不是集合的迭代器
var 以保存我要查找的文件名的结果。另外，为了在找到我的文件后停止循环，我需要将 while 循环包装在 breakable 中，然后在找到文件后我可以 break。
方法 1
对于 pyspark 用户，我已经翻译了 Michael Spector 的 answer（由您决定使用它是否是一个好主意）：
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 现在保存在my-bucket/my-prefix 找到的所有文件密钥
方法 2 这是我发现的另一种选择（hat tip 到 @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()
方法 3（不完整！）
上述两种方法不使用将应用于分布式读取的 Spark 并行机制。不过，这种逻辑看起来很私密。见parallelListLeafFileshere。我还没有找到一种方法来强制 pyspark 在 s3 上分发 ls 而不读取文件内容。我尝试使用 py4j 实例化 InMemoryFileIndex，但无法正确使用咒语。如果有人想从这里取走，这是我目前所拥有的：
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
)
【讨论】：