HDFS: Interfaz Java (FileSystem)

Como ya comenté en una entrada anterior (Hadoop Distributed File System (HDFS)), para acceder a HDFS lo podemos hacer de dos formas, por un lado y como vimos en esa misma entrada (con algunos ejemplos) utilizando un terminal y a través de los comandos propios de hadoop y por otro lado a través de la API de Java, que es la parte que vamos a ver ahora.

Para interactuar con HDFS con Java, se hace a través de la clase FileSystem.
El desarrollo es en Java y los ejecutamos desde línea de comando tras haber arrancado los demonios Hadoop (/bin/start-all.sh).

Lo primero que he hecho es crear una carpeta pruebas en mi sistema de ficheros HDFS:
      hadoop fs -mkdir pruebas

Algunos ejemplos que también podéis descargar en este enlace son:

• Copiar un fichero o carpeta del sistema de ficheros local a HDFS

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

public class LocalToHdfs {
 public static void main(String[] args) throws Exception{
  Configuration conf = new Configuration();
  FileSystem hdfs = FileSystem.get(conf);
  Path sourcePath = new Path(args[0]);
  Path destPath = new Path(args[1]);
  hdfs.copyFromLocalFile(sourcePath, destPath);
 }
}

Y este es el resultado en el que copio desde mi carpeta local training el fichero score.txt a mi carpeta pruebas en HDFS:

$ hadoop jar training/JavaHdfs.jar LocalToHdfs training/score.txt /user/elena/pruebas 
$ hadoop fs -ls pruebas 
Found 1 items -rw-r--r--  1 elena supergroup  363 2013-02-11 19:31 /user/elena/pruebas/score.txt

• Crear un fichero HDFS

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

public class CreateFileHdfs {

 public static void main(String[] args) throws Exception{
  String contentFile = "Este es el contenido del fichero\n";
  byte[] texto = contentFile.getBytes();
  Configuration conf = new Configuration();
  FileSystem hdfs = FileSystem.get(conf);
  Path path = new Path(args[0]+"/nombreFichero");
  FSDataOutputStream outputStream = hdfs.create(path);
  outputStream.write(texto, 0, texto.length);
 }
}

En el código Java tengo puesto el nombre del fichero y paso como argumento la carpeta donde se debe crear el fichero.

$ hadoop jar training/JavaHdfs.jar CreateFileHdfs pruebas 
$ hadoop fs -ls pruebas 
Found 2 items 
-rw-r--r--  1 elena supergroup  32 2013-02-11 19:34 /user/elena/pruebas/nombreFichero 
-rw-r--r--  1 elena supergroup  363 2013-02-11 19:31 /user/elena/pruebas/score.txt 
$ hadoop fs -cat pruebas/nombreFichero 
Este es el contenido del fichero

• Leer el contenido de un fichero HDFS

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

public class ReadFileHdfs {

 public static void main(String[] args) throws Exception {
  Configuration conf = new Configuration();
  FileSystem hdfs = FileSystem.get(conf);
  Path path = new Path(args[0]+"/nombreFichero");
  FSDataInputStream inputStream = hdfs.open(path);
  IOUtils.copyBytes(inputStream, System.out, conf, true); 
 }
}

$ hadoop jar training/JavaHdfs.jar ReadFileHdfs pruebas 
Este es el contenido del fichero 

• Borrar un fichero HDFS

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

public class DeleteFileHdfs {

 public static void main(String[] args) throws Exception{
  Configuration conf = new Configuration();
  FileSystem hdfs = FileSystem.get(conf);
  Path path = new Path(args[0]+"/nombreFichero");
  hdfs.delete(path, false);
 }
}

$ hadoop jar training/JavaHdfs.jar DeleteFileHdfs pruebas 
$ hadoop fs -ls pruebas 
Found 1 items 
-rw-r--r--  1 elena supergroup  363 2013-02-11 19:31 /user/elena/pruebas/score.txt 

Alta Disponibilidad

En la última versión de Hadoop, la CDH4, se ha puesto en marcha la Alta Disponibilidad.

Hasta ahora, el NameNode era un punto de fallo crítico (SPOF en inglés) de nuestro sistema Hadoop. Si el NameNode falla, todos los clientes, incluído los Job MapReduce se inutilizan.
El administrador tendría que encargarse de crear un nuevo NameNode a partir de una réplica de los metadata y configurar los clientes y DataNodes para utilizar este nuevo NameNode, más todas las tareas que hay por detrás hasta que el NameNode está disponible para funcionar.
Hasta que no se hace todo eso, el sistema se encuentra bloqueado.

Un tiempo de recuperación largo puede ser un grabe problema en ciertas situaciones.

Como hemos visto, en un cluster Hadoop disponemos de un NameNode y de un Secondary Namenode, pero si la Alta Disponibilidad está activada, en vez de disponer de estos dos elementos dispondríamos de un Active NameNode y de un Standby NameNode que estarían haciendo las mismas funciones de principal y secundario y el Standby realizando las tareas que hemos visto que ejerce el Secondary.
Pero en este caso, si el Active NameNode falla, el Standby NameNode se activa automáticamente (para esto se necesita tener un cluster Zookeeper) y empezaría a funcionar como el NameNode ya que el Standby NameNode siempre tiene en memoria la misma información de metadata que el NameNode activo (en caso de alta disponibilidad, los DataNodes envían sus informes tanto al NameNode activo como al Standby).

Hadoop Distributed File System (HDFS)

El HDFS es el sistema de archivos distribuído de Hadoop.

Los datos se dividen en bloques de 64MB o 128MB, estos bloques se replican en un número configurable de veces (por defecto 3) y se distribuyen en distintos nodos del clúster. Esto es porque HDFS supone que algún componente del sistema puede fallar, así que siempre va a tener disponible alguna de las copias. Por otro lado, tienen que existir tantos bloques como hayamos configurado, por eso, si HDFS detecta que un componente no funciona, se encargará de saber qué bloques estaban en ese componente y replicarlos por el resto del clúster.

Es importante saber que cuando un fichero se divide en bloques, si un bloque queda "incompleto" (respecto del tamaño del bloque establecido), ese espacio nunca se va a rellenar ni utilizar con datos de otro fichero.

Este sistema de archivos está más pensado para tener un número de ficheros razonable (no más de algunos millones) de un tamaño bastante grande (del orden del tamaño de bloque HDFS o más grande aún). Es decir, con HDFS es mejor tener menos ficheros, pero más grandes (un ítem de metadatos ocupa entre 150 y 200 bytes de memoria RAM en el NameNode).
También es más óptimo para lecturas en streaming.

Está escrito en Java. Y los bloques de datos HDFS se escriben en ficheros (dentro del sistema de archivos estándar del servidor, por ejemplo ext4) en unos directorios específicos de los nodos esclavos (se encargaría el administrador) sólo una vez y no se les puede añadir más información.

Para que HDFS pueda ser un sistema de archivos distribuído, se ha de separar los datos de los metadatos. Para esto utiliza dos tipos de nodos: el NameNode y los DataNodes de los que ya hablaré más detalle en otra entrada.
Resumidamente el NameNode contiene los metadatos, es quien se encarga de dividir y distribuir los bloques por los DataNodes del clúster. Cuando un cliente quiere hacer una lectura de los datos, primero va a preguntar al NameNode, que es quien sabe dónde está cada bloque, y luego va a leer los datos bloque a bloque y en cada uno de los nodos que los contienen.

Características:

• Almacenamiento redundante
• Tolerancia a fallos
• Mecanismo de checksums
• Política de una escritura (write-once), preparado para muchas lecturas.
• Escalabilidad horizontal
• No implementa POSIX

Bloques HDFS

Un cliente puede acceder a HDFS a través de un terminal utilizando los comandos propios de HDFS o vía la API de Java a través de una aplicación.
Y aunque yo o mucha gente que lea el blog sean perfiles de desarrollo, es muy importante saber moverse y tener unos conocimientos básicos de cómo funciona este sistema de ficheros a través del terminal.
Para acceder a través de la línea de comandos se hace con el comando:

     $ hadoop fs

Por sí solo este comando no hace nada (sólo muestra la ayuda), luego existen múltiples comandos que ayudarán al usuario a hacer ciertas interacciones.

Cuando accedemos a través del terminal hay que tener muy claro cuál es nuestro sistema de ficheros tradicional y cuándo estamos en HDFS.

Algunos de estos comandos asociados a hadoop son:

• Comando encargado de copiar el archivo ficheroLocal.txt que se encuentra en el directorio actual de mi sistema de ficheros local a HDFS, el archivo se llamará ficheroHDFS.txt, pero se encontrará en el directorio HDFS configurado por nuestro administrador (nuestra carpeta de usuario en HDFS):

          $ hadoop fs -put ficheroLocal.txt ficheroHDFS.txt

• Ahora el comando encargado de copiar un fichero desde HDFS a nuestro sistema de ficheros local:

                    $ hadoop fs -get ficheroHDFS.txt ficheroLocal.txt

• Si queremos listar el contenido de nuestro directorio HDFS (nuestra carpeta de usuario):

          $ hadoop fs -ls

• Para mostrar el contenido de un fichero en nuestra carpeta de usuario HDFS:

                    $ hadoop fs -cat ficheroHDFS.txt

• Queremos crear un directorio en HDFS (en nuestra carpeta de usuario)

                    $ hadoop fs -mkdir miDirectorio

• Queremos borrar un directorio y todo su contenido:

                    $ hadoop fs -rm -r miDirectorio


Sobre cómo acceder con la API de Java, sería a través del FileSystem API, pero es una parte extensa y ya lo mostraré en otra entrada más adelante.

Desventajas de Hadoop

Como hemos visto en entradas anteriores, Hadoop es una tecnología que mejora considerablemente respecto de los típicos sistemas distribuídos.
Eso no quita que tenga una serie de desventajas y que si en algún momento estamos estudiando si usarlo o no, hay que tenerlas en cuenta.

En lo que respecta al HDFS:

• Latencia para el acceso a datos: HDFS está orientado a procesos batch y operaciones en streaming. Por lo tanto, la latencia de cualquier operación IO no ha sido optimizada y sistemas de archivos tradicionales (como ext4, XFS...) suelen ser más rápidos en estos aspectos.
• Cantidades grandes de ficheros pequeños: El límite del número de ficheros en este sistema está limitado por la memoria del NameNode, que es en su RAM donde se encuentran los metadata. Cada fichero, directorio y bloque ocupa un tamaño de entre 150 y 200 bytes en los metadata, lo que quiere decir, que si hay millones de ficheros pequeños va a ocupar mucho más espacio en la RAM que si tenemos menos cantidad de ficheros de gran tamaño (recomendable 100 MB o más).
• Escribe una vez, lee varias: En HDFS los ficheros solo se pueden escribir una vez (aunque HDFS se ha mejorado con el modo "append", Cloudera no suele recomendarlo porque no lo considera estable).
• No se puede acceder con los comandos tradicionales de Linux (ls, cat, vim...). Esto complica mucho la integración con otras herramientas comerciales (como sistemas de backup, por ejemplo). Y aunque exista "HDFS fuse" para montar HDFS como cualquier otro sistema de archivo Linux, esta solución no ofrece un buen rendimiento.

En lo que respecta a MapReduce:

• Es muy difícil de depurar: Al procesarse el programa en los nodos donde se encuentran los bloques de datos, no es fácil encontrar los fallos de código. Tampoco es conveniente utilizar funciones de escritura de logs en el código ya que eso podría suponer un gran aumento en la ejecución de procesos MapReduce.
• No todos los algoritmos se pueden escribir con el paradigma MapReduce. Por ejemplo, el famoso algoritmo de grafos Dijkstra, al necesitar un procesamiento secuencial, no se puede escribir en MapReduce.
• Latencia: cualquier job MapReduce suele tardar por lo menos 10 segundos. Por lo tanto, si el volumen de información a tratar es pequeño, es posible que Hadoop no sea la solución más rápida.