Cómo cargar un archivo local en sc.textFile, en lugar de HDFS

100

Estoy siguiendo el tutorial de Great Spark

así que estoy tratando a las 46m: 00s de cargar el README.mdpero no lo que estoy haciendo es esto:

$ sudo docker run -i -t -h sandbox sequenceiq/spark:1.1.0 /etc/bootstrap.sh -bash
bash-4.1# cd /usr/local/spark-1.1.0-bin-hadoop2.4
bash-4.1# ls README.md
README.md
bash-4.1# ./bin/spark-shell
scala> val f = sc.textFile("README.md")
14/12/04 12:11:14 INFO storage.MemoryStore: ensureFreeSpace(164073) called with curMem=0, maxMem=278302556
14/12/04 12:11:14 INFO storage.MemoryStore: Block broadcast_0 stored as values in memory (estimated size 160.2 KB, free 265.3 MB)
f: org.apache.spark.rdd.RDD[String] = README.md MappedRDD[1] at textFile at <console>:12
scala> val wc = f.flatMap(l => l.split(" ")).map(word => (word, 1)).reduceByKey(_ + _)
org.apache.hadoop.mapred.InvalidInputException: Input path does not exist: hdfs://sandbox:9000/user/root/README.md
    at org.apache.hadoop.mapred.FileInputFormat.singleThreadedListStatus(FileInputFormat.java:285)

¿cómo puedo cargar eso README.md?

Jas
fuente

Respuestas:

177

Intente especificar explícitamente sc.textFile("file:///path to the file/"). El error ocurre cuando se configura el entorno Hadoop.

SparkContext.textFile llama internamente org.apache.hadoop.mapred.FileInputFormat.getSplits, que a su vez usa org.apache.hadoop.fs.getDefaultUrisi el esquema está ausente. Este método lee el parámetro "fs.defaultFS" de Hadoop conf. Si establece la variable de entorno HADOOP_CONF_DIR, el parámetro generalmente se establece como "hdfs: // ..."; de lo contrario, "archivo: //".

suztomo
fuente
¿Sabes cómo hacer esto con Java? No veo un método. Me resulta muy frustrante que no haya una manera fácil de dar una ruta para cargar un archivo desde un sistema de archivos simple.
Brad Ellis
respondiéndome a mí mismo. Hay un modificador --file que pasa con el Spark-Submit. Por lo tanto, la ruta del archivo puede estar codificada o, sin embargo, su configuración está configurada para la aplicación, pero también indica esa ruta. cuando envíe para que los ejecutores puedan ver la ruta.
Brad Ellis
24

La respuesta de Gonbe es excelente. Pero aún quiero mencionar que file:///= ~/../../, no $SPARK_HOME. Espero que esto pueda ahorrar algo de tiempo para novatos como yo.

zaxliu
fuente
4
file:///es la carpeta raíz del sistema de archivos como la ve la JVM en ejecución, no dos niveles por encima de la carpeta de inicio. El formato de URI como se especifica en RFC 8089 es file://hostname/absolute/path. En el caso local, el hostnamecomponente (autoridad) está vacío.
Hristo Iliev
17

Si bien Spark admite la carga de archivos desde el sistema de archivos local, requiere que los archivos estén disponibles en la misma ruta en todos los nodos de su clúster.

Algunos sistemas de archivos de red, como NFS, AFS y la capa NFS de MapR, están expuestos al usuario como un sistema de archivos normal.

Si sus datos ya están en uno de estos sistemas, puede usarlos como entrada simplemente especificando un archivo: // ruta; Spark lo manejará siempre que el sistema de archivos esté montado en la misma ruta en cada nodo. Cada nodo debe tener la misma ruta

 rdd = sc.textFile("file:///path/to/file")

Si su archivo aún no está en todos los nodos del clúster, puede cargarlo localmente en el controlador sin pasar por Spark y luego llamar a paralelizar para distribuir el contenido a los trabajadores.

Tenga cuidado de poner file: // al principio y el uso de "/" o "\" según el sistema operativo.

Aklank Jain
fuente
1
¿Hay alguna manera de que Spark copie automáticamente los datos de su directorio $ SPARK_HOME a todos los nodos informáticos? ¿O necesitas hacerlo manualmente?
Matthias
¿Dónde está el código fuente de Spark manejando diferentes formatos de sistema de archivos?
Saher Ahwal
12

Solo necesita especificar la ruta del archivo como "archivo: /// directorio / archivo"

ejemplo:

val textFile = sc.textFile("file:///usr/local/spark/README.md")
Hamdi Charef
fuente
12

Atención:

Asegúrese de ejecutar Spark en modo local cuando cargue datos desde local ( sc.textFile("file:///path to the file/")) o obtendrá un error como este Caused by: java.io.FileNotFoundException: File file:/data/sparkjob/config2.properties does not exist. Porque los ejecutores que se ejecutan en diferentes trabajadores no encontrarán este archivo en su ruta local.

Matiji66
fuente
11

Si el archivo está ubicado en su nodo maestro de Spark (por ejemplo, en caso de usar AWS EMR), primero inicie Spark-Shell en modo local.

$ spark-shell --master=local
scala> val df = spark.read.json("file:///usr/lib/spark/examples/src/main/resources/people.json")
df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]

scala> df.show()
+----+-------+
| age|   name|
+----+-------+
|null|Michael|
|  30|   Andy|
|  19| Justin|
+----+-------+

Alternativamente, primero puede copiar el archivo a HDFS desde el sistema de archivos local y luego iniciar Spark en su modo predeterminado (por ejemplo, YARN en caso de usar AWS EMR) para leer el archivo directamente.

$ hdfs dfs -mkdir -p /hdfs/spark/examples
$ hadoop fs -put /usr/lib/spark/examples/src/main/resources/people.json /hdfs/spark/examples
$ hadoop fs -ls /hdfs/spark/examples
Found 1 items
-rw-r--r--   1 hadoop hadoop         73 2017-05-01 00:49 /hdfs/spark/examples/people.json

$ spark-shell
scala> val df = spark.read.json("/hdfs/spark/examples/people.json")
df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]

scala> df.show()
+----+-------+
| age|   name|
+----+-------+
|null|Michael|
|  30|   Andy|
|  19| Justin|
+----+-------+
Joarder Kamal
fuente
9

Tengo un archivo llamado NewsArticle.txt en mi escritorio.

En Spark, escribí:

val textFile= sc.textFile(“file:///C:/Users/582767/Desktop/NewsArticle.txt”)

Necesitaba cambiar todo el carácter \ a / para la ruta del archivo.

Para probar si funcionó, escribí:

textFile.foreach(println)

Estoy ejecutando Windows 7 y no tengo Hadoop instalado.

Gene
fuente
5

Esto se ha discutido en la lista de correo de Spark, y consulte este correo .

Debe usar hadoop fs -put <localsrc> ... <dst>copiar el archivo en hdfs:

${HADOOP_COMMON_HOME}/bin/hadoop fs -put /path/to/README.md README.md
Nan Xiao
fuente
5

Esto me ha sucedido con Spark 2.3 con Hadoop también instalado en el directorio de inicio del usuario común "hadoop". Dado que tanto Spark como Hadoop se instalaron en el mismo directorio común, Spark por defecto considera el esquema como hdfsy comienza a buscar los archivos de entrada bajo hdfs como se especifica fs.defaultFSen Hadoop's core-site.xml. En tales casos, necesitamos especificar explícitamente el esquema como file:///<absoloute path to file>.

Binita Bharati
fuente
0

Esta es la solución para este error que estaba obteniendo en el clúster Spark que está alojado en Azure en un clúster de Windows:

Cargue el archivo HVAC.csv sin procesar, analícelo usando la función

data = sc.textFile("wasb:///HdiSamples/SensorSampleData/hvac/HVAC.csv")

Usamos (wasb: ///) para permitir que Hadoop acceda al archivo de almacenamiento de blog azul y las tres barras diagonales son una referencia relativa a la carpeta del contenedor del nodo en ejecución.

Por ejemplo: si la ruta de su archivo en el Explorador de archivos en el panel del clúster de Spark es:

sflcc1 \ sflccspark1 \ HdiSamples \ SensorSampleData \ hvac

Entonces, para describir la ruta es la siguiente: sflcc1: es el nombre de la cuenta de almacenamiento. sflccspark: es el nombre del nodo del clúster.

Por tanto, nos referimos al nombre del nodo del clúster actual con las tres barras diagonales relativas.

Espero que esto ayude.

Mostafa
fuente
0

Si está intentando leer el archivo HDFS. intentando establecer la ruta en SparkConf

 val conf = new SparkConf().setMaster("local[*]").setAppName("HDFSFileReader")
 conf.set("fs.defaultFS", "hdfs://hostname:9000")
Viyaan Jhiingade
fuente
Agregue una sangría de 4 espacios / tabulación a su código para que se formatee como código. Saludos cordiales
YakovL
0

No es necesario utilizar sc.textFile (...) para convertir archivos locales en marcos de datos. Una de las opciones es leer un archivo local línea por línea y luego transformarlo en Spark Dataset. Aquí hay un ejemplo para una máquina Windows en Java:

StructType schemata = DataTypes.createStructType(
            new StructField[]{
                    createStructField("COL1", StringType, false),
                    createStructField("COL2", StringType, false),
                    ...
            }
    );

String separator = ";";
String filePath = "C:\\work\\myProj\\myFile.csv";
SparkContext sparkContext = new SparkContext(new SparkConf().setAppName("MyApp").setMaster("local"));
JavaSparkContext jsc = new JavaSparkContext (sparkContext );
SQLContext sqlContext = SQLContext.getOrCreate(sparkContext );

List<String[]> result = new ArrayList<>();
try (BufferedReader br = new BufferedReader(new FileReader(filePath))) {
    String line;
    while ((line = br.readLine()) != null) {
      String[] vals = line.split(separator);
      result.add(vals);
    }
 } catch (Exception ex) {
       System.out.println(ex.getMessage());
       throw new RuntimeException(ex);
  }
  JavaRDD<String[]> jRdd = jsc.parallelize(result);
  JavaRDD<Row> jRowRdd = jRdd .map(RowFactory::create);
  Dataset<Row> data = sqlContext.createDataFrame(jRowRdd, schemata);

Ahora puedes usar dataframe dataen tu código.

Andrushenko Alexander
fuente
0

Intenté lo siguiente y funcionó desde mi sistema de archivos local ... Básicamente, Spark puede leer desde la ruta local, HDFS y AWS S3

listrdd=sc.textFile("file:////home/cloudera/Downloads/master-data/retail_db/products")
BigData-Guru
fuente
-6

tratar

val f = sc.textFile("./README.md")
Soumya Simanta
fuente
scala> val f = sc.textFile("./README.md") 14/12/04 12:54:33 INFO storage.MemoryStore: ensureFreeSpace(81443) called with curMem=164073, maxMem=278302556 14/12/04 12:54:33 INFO storage.MemoryStore: Block broadcast_1 stored as values in memory (estimated size 79.5 KB, free 265.2 MB) f: org.apache.spark.rdd.RDD[String] = ./README.md MappedRDD[5] at textFile at <console>:12 scala> val wc = f.flatMap(l => l.split(" ")).map(word => (word, 1)).reduceByKey(_ + _) org.apache.hadoop.mapred.InvalidInputException: Input path does not exist: hdfs://sandbox:9000/user/root/README.md at
Jas
¿Puedes hacer un pwden el bash shell?bash-4.1#
Soumya Simanta
bash-4.1 # pwd /usr/local/spark-1.1.0-bin-hadoop2.4
Jas
Esto funciona para mí en Spark sin hadoop / hdfs. Sin embargo, no parece estar funcionando para el OP, ya que les dio un volcado de error.
Paul