06 - Lectura y escritura de ficheros.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![RDD key pair](media/07.spark_statestore_files.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "# 06 - Lectura y escritura de ficheros\n",
    "--------------\n",
    "\n",
    "### Sistemas de ficheros soportados\n",
    "-   Igual que Hadoop, Spark soporta diferentes filesystems: local, HDFS, Amazon S3\n",
    "\n",
    "    -   En general, soporta cualquier fuente de datos que se pueda leer con Hadoop\n",
    "\n",
    "-   También, acceso a bases de datos relacionales o noSQL\n",
    "\n",
    "    -   MySQL, Postgres, etc. mediante JDBC\n",
    "    -   Apache Hive, HBase, Cassandra o Elasticsearch\n",
    "\n",
    "### Formatos de fichero soportados\n",
    "\n",
    "-   Spark puede acceder a diferentes tipos de ficheros:\n",
    "\n",
    "    -   Texto plano, CSV, ficheros sequence, JSON, *protocol buffers* y *object files*\n",
    "        -   Soporta ficheros comprimidos\n",
    "    -   Veremos el acceso a algunos tipos en esta clase, y dejaremos otros para más adelante "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Requirement already satisfied: pyspark in /home/xergioalex/Envs/tensorflow/lib/python3.5/site-packages\n",
      "Requirement already satisfied: py4j==0.10.7 in /home/xergioalex/Envs/tensorflow/lib/python3.5/site-packages (from pyspark)\n",
      "\u001b[33mYou are using pip version 9.0.1, however version 18.0 is available.\n",
      "You should consider upgrading via the 'pip install --upgrade pip' command.\u001b[0m\n"
     ]
    }
   ],
   "source": [
    "!pip install pyspark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create apache spark context\n",
    "from pyspark import SparkContext\n",
    "sc = SparkContext(master=\"local\", appName=\"Mi app\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Stop apache spark context\n",
    "sc.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Ejemplos con ficheros de texto\n",
    "\n",
    "En el directorio `data/libros` hay un conjunto de ficheros de texto comprimidos."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "pg14329.txt.gz\tpg17013.txt.gz\tpg2000.txt.gz\tpg25807.txt.gz\tpg7109.txt.gz\r\n",
      "pg1619.txt.gz\tpg17073.txt.gz\tpg24536.txt.gz\tpg32315.txt.gz\tpg8870.txt.gz\r\n",
      "pg16625.txt.gz\tpg18005.txt.gz\tpg25640.txt.gz\tpg5201.txt.gz\tpg9980.txt.gz\r\n"
     ]
    }
   ],
   "source": [
    "# Ficheros de entrada\n",
    "!ls data/libros"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Funciones de lectura y escritura con ficheros de texto\n",
    "\n",
    "\n",
    "- `sc.textFile(nombrefichero/directorio)` Crea un RDD a partir las líneas de uno o varios ficheros de texto\n",
    "    - Si se especifica un directorio, se leen todos los ficheros del mismo, creando una partición por fichero\n",
    "    - Los ficheros pueden estar comprimidos, en diferentes formatos (gz, bz2,...)\n",
    "    - Pueden especificarse comodines en los nombres de los ficheros\n",
    "- `sc.wholeTextFiles(nombrefichero/directorio)` Lee ficheros y devuelve un RDD clave/valor\n",
    "    - clave: path completo al fichero\n",
    "    - valor: el texto completo del fichero\n",
    "- `rdd.saveAsTextFile(directorio_salida)` Almacena el RDD en formato texto en el directorio indicado\n",
    "    - Crea un fichero por partición del rdd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Número de particiones del RDD lineas = 15\n"
     ]
    }
   ],
   "source": [
    "# Lee todos los ficheros del directorio\n",
    "# y crea un RDD con las líneas\n",
    "lineas = sc.textFile(\"data/libros\")\n",
    "\n",
    "# Se crea una partición por fichero de entrada\n",
    "print(\"Número de particiones del RDD lineas = {0}\".format(lineas.getNumPartitions()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Número de particiones del RDD palabras = 15\n"
     ]
    }
   ],
   "source": [
    "# Obtén las palabras usando el método split (split usa un espacio como delimitador por defecto)\n",
    "palabras = lineas.flatMap(lambda x: x.split())\n",
    "print(\"Número de particiones del RDD palabras = {0}\".format(palabras.getNumPartitions()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Número de particiones del RDD palabras2 = 4\n"
     ]
    }
   ],
   "source": [
    "# Reparticiono el RDD en 4 particiones       \n",
    "palabras2 = palabras.coalesce(4)\n",
    "print(\"Número de particiones del RDD palabras2 = {0}\".format(palabras2.getNumPartitions()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['en', 'labio', '_pescas_', 'ahumado', 'amena.', 'mucho', 'inocente.', 'sosona', 'Roma,', 'O']\n"
     ]
    }
   ],
   "source": [
    "# Toma una muestra aleatoria de palabras\n",
    "print(palabras2.takeSample(False, 10))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Número de particiones del RDD prdd = 1\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Lee los ficheros y devuelve un RDD clave/valor\n",
    "# clave->nombre fichero, valor->fichero completo\n",
    "prdd = sc.wholeTextFiles(\"data/libros/p*.gz\")\n",
    "print(\"Número de particiones del RDD prdd = {0}\\n\".format(prdd.getNumPartitions()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "El fichero pg14329.txt.gz tiene 183777 palabras\n",
      "El fichero pg1619.txt.gz  tiene 109878 palabras\n",
      "El fichero pg16625.txt.gz tiene 170900 palabras\n",
      "El fichero pg17013.txt.gz tiene 396086 palabras\n",
      "El fichero pg17073.txt.gz tiene 309473 palabras\n",
      "El fichero pg18005.txt.gz tiene  86446 palabras\n",
      "El fichero pg2000.txt.gz  tiene 384258 palabras\n",
      "El fichero pg24536.txt.gz tiene 134016 palabras\n",
      "El fichero pg25640.txt.gz tiene 207338 palabras\n",
      "El fichero pg25807.txt.gz tiene  15014 palabras\n",
      "El fichero pg32315.txt.gz tiene  46142 palabras\n",
      "El fichero pg5201.txt.gz  tiene  49441 palabras\n",
      "El fichero pg7109.txt.gz  tiene  35037 palabras\n",
      "El fichero pg8870.txt.gz  tiene  54348 palabras\n",
      "El fichero pg9980.txt.gz  tiene  34014 palabras\n"
     ]
    }
   ],
   "source": [
    "# Obtiene un lista clave/valor\n",
    "# clave->nombre fichero, valor->numero de palabras\n",
    "lista = prdd.mapValues(lambda x: len(x.split())).collect()\n",
    "\n",
    "for libro in lista:\n",
    "    print(\"El fichero {0:14s} tiene {1:6d} palabras\".format(libro[0].split(\"/\")[-1], libro[1]))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Ficheros Sequence\n",
    "Ficheros clave/valor usados en Hadoop\n",
    "\n",
    "-   Sus elementos implementan la interfaz [`Writable`](https://hadoop.apache.org/docs/stable/api/org/apache/hadoop/io/Writable.html)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([(\"a\",2), (\"b\",5), (\"a\",8)], 2)\n",
    "\n",
    "# Salvamos el RDD clave valor como fichero de secuencias\n",
    "rdd.saveAsSequenceFile(\"file:///tmp/sequenceoutdir2\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Contenido del RDD [('b', 5), ('a', 8), ('a', 2)]\n"
     ]
    }
   ],
   "source": [
    "# Lo leemos en otro RDD\n",
    "rdd2 = sc.sequenceFile(\"file:///tmp/sequenceoutdir2\", \n",
    "                       \"org.apache.hadoop.io.Text\", \n",
    "                       \"org.apache.hadoop.io.IntWritable\")\n",
    "                       \n",
    "print(\"Contenido del RDD {0}\".format(rdd2.collect()))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Formatos de entrada/salida de Hadoop\n",
    "Spark puede interactuar con cualquier formato de fichero soportado por Hadoop \n",
    "- Soporta las APIs “vieja” y “nueva”\n",
    "- Permite acceder a otros tipos de almacenamiento (no fichero), p.e. HBase o MongoDB, a través de `saveAsHadoopDataSet` y/o `saveAsNewAPIHadoopDataSet`\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Salvamos el RDD clave/valor como fichero de texto Hadoop (TextOutputFormat)\n",
    "rdd.saveAsNewAPIHadoopFile(\"file:///tmp/hadoopfileoutdir\", \n",
    "                            \"org.apache.hadoop.mapreduce.lib.output.TextOutputFormat\",\n",
    "                            \"org.apache.hadoop.io.Text\",\n",
    "                            \"org.apache.hadoop.io.IntWritable\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Directorio de salida\n",
      "total 8\n",
      "-rw-r--r-- 1 xergioalex xergioalex 0 Sep 23 18:28 _SUCCESS\n",
      "-rw-r--r-- 1 xergioalex xergioalex 4 Sep 23 18:28 part-r-00000\n",
      "-rw-r--r-- 1 xergioalex xergioalex 8 Sep 23 18:28 part-r-00001\n",
      "b\t5\n",
      "a\t8\n"
     ]
    }
   ],
   "source": [
    "!echo 'Directorio de salida'\n",
    "!ls -l /tmp/hadoopfileoutdir\n",
    "!cat /tmp/hadoopfileoutdir/part-r-00001"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Contenido del RDD [('a', '2'), ('b', '5'), ('a', '8')]\n"
     ]
    }
   ],
   "source": [
    "# Lo leemos como fichero clave-valor Hadoop (KeyValueTextInputFormat)\n",
    "rdd3 = sc.newAPIHadoopFile(\"file:///tmp/hadoopfileoutdir\", \n",
    "                          \"org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat\",\n",
    "                          \"org.apache.hadoop.io.Text\",\n",
    "                          \"org.apache.hadoop.io.IntWritable\")\n",
    "                          \n",
    "print(\"Contenido del RDD {0}\".format(rdd3.collect()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.5.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}