hadoop - retainedjobs - spark port maxretries
Spark iterate directorio HDFS (6)
@Tagar no dijo cómo conectar hdfs remotos, pero esta respuesta lo hizo:
URI = sc._gateway.jvm.java.net.URI
Path = sc._gateway.jvm.org.apache.hadoop.fs.Path
FileSystem = sc._gateway.jvm.org.apache.hadoop.fs.FileSystem
Configuration = sc._gateway.jvm.org.apache.hadoop.conf.Configuration
fs = FileSystem.get(URI("hdfs://somehost:8020"), Configuration())
status = fs.listStatus(Path(''/some_dir/yet_another_one_dir/''))
for fileStatus in status:
print(fileStatus.getPath())
Tengo un directorio de directorios en HDFS, y quiero iterar sobre los directorios. ¿Hay alguna manera fácil de hacer esto con Spark usando el objeto SparkContext?
Aquí está la versión de PySpark si alguien está interesado:
hadoop = sc._jvm.org.apache.hadoop
fs = hadoop.fs.FileSystem
conf = hadoop.conf.Configuration()
path = hadoop.fs.Path(''/hivewarehouse/disc_mrt.db/unified_fact/'')
for f in fs.get(conf).listStatus(path):
print f.getPath()
En este caso particular, obtengo una lista de todos los archivos que conforman la tabla Hive de disc_mrt.unified_fact.
A continuación se describen otros métodos del objeto FileStatus, como getLen () para obtener el tamaño del archivo:
También puedes probar con el estado globStatus
val listStatus = org.apache.hadoop.fs.FileSystem.get(new URI(url), sc.hadoopConfiguration).globStatus(new org.apache.hadoop.fs.Path(url))
for (urlStatus <- listStatus) {
println("urlStatus get Path:"+urlStatus.getPath())
}
esto hizo el trabajo por mi
FileSystem.get(new URI("hdfs://HAservice:9000"), sc.hadoopConfiguration).listStatus( new Path("/tmp/")).foreach( x => println(x.getPath ))
Puede utilizar org.apache.hadoop.fs.FileSystem
. Específicamente, FileSystem.listFiles([path], true)
Y con Spark ...
FileSystem.get(sc.hadoopConfiguration()).listFiles(..., true)
Editar
Vale la pena señalar que una buena práctica es obtener el sistema de FileSystem
asociado con el esquema de la Path
.
path.getFileSystem(sc.hadoopConfiguration).listFiles(path, true)
import org.apache.hadoop.fs.{FileSystem,Path}
FileSystem.get( sc.hadoopConfiguration ).listStatus( new Path("hdfs:///tmp")).foreach( x => println(x.getPath ))
Esto funcionó para mí.
Spark versión 1.5.0-cdh5.5.2