08 - Spark SQL.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![RDD key pair](media/15.spark_sql.jpg)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 08 - Spark SQL\n",
    "\n",
    "-   Interfaz para trabajar con datos estructurados y semiestructurados\n",
    "\n",
    "\n",
    "-   Capacidades principales\n",
    "    -   Lee datos de una gran variedad de fuentes: RDDs, ficheros JSON, Hive, HDFS, Parquet…\n",
    "    -   Permite consultas SQL, tanto desde programas Spark como externas usando conectores estándar (JDBC/ODBC)\n",
    "    -   Integra SQL y código Spark normal **(en Python/Java/Scala)**\n",
    "\n",
    "\n",
    "-   Contexto SQLContext: punto de entrada (equivalente al SparkContext)"
   ]
  },
  {
   "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": 11,
   "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": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Stop apache spark context\n",
    "sc.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Elemento básico: DataFrames\n",
    "\n",
    "-   Colección distribuida de datos organizada en columnas con nombre\n",
    "    - Conceptualmente equivalente a una tabla en una BD o a un dataframe en R o Python Pandas\n",
    "    - Al igual que los RDDs son inmutables y lazy\n",
    "    - Desarrollados dentro de Spark SQL\n",
    "        - Permite acceder a los datos mediante consultas SQL\n",
    "        - Sustitutos de los RDDs en general\n",
    "\n",
    "\n",
    "-   `DataSet`: nuevo tipo de datos añadido en Spark 1.6\n",
    "    -   Intenta proporcionar los beneficios de los RDDs con las optimizaciones que proporciona el motor de ejecución [Tungsten](https://databricks.com/blog/2015/04/28/project-tungsten-bringing-spark-closer-to-bare-metal.html \"Project Tungsten: Bringing Apache Spark Closer to Bare Metal\") de Spark SQL.\n",
    "    -   Sólo disponible en Scala y Java\n",
    "    -   En [Java](http://spark.apache.org/docs/latest/api/java/index.html \"Interface Row\") y [Scala](http://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.Row \"trait Row extends Serializable\"), un DataFrame es un DataSet de objetos de tipo Row"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Mejora de rendimiento\n",
    "\n",
    "- Spark SQL con DataFrames y DataSets se aprovecha del uso de datos con estructura para optimizar el rendimiento utilizando el optimizador de consultas [Catalyst](https://databricks.com/blog/2015/04/13/deep-dive-into-spark-sqls-catalyst-optimizer.html \"Deep Dive into Spark SQL’s Catalyst Optimizer\")  y el motor de ejecución [Tungsten](https://databricks.com/blog/2015/04/28/project-tungsten-bringing-spark-closer-to-bare-metal.html \"Project Tungsten: Bringing Apache Spark Closer to Bare Metal\").\n",
    "\n",
    "![Mejora de rendimiento](media/14.performance.png)\n",
    "\n",
    "Fuente: [Recent performance improvements in Apache Spark: SQL, Python, DataFrames, and More](https://databricks.com/blog/2015/04/24/recent-performance-improvements-in-apache-spark-sql-python-dataframes-and-more.html \"Recent performance improvements in Apache Spark: SQL, Python, DataFrames, and More\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Creación de DataFrames\n",
    "Varias formas:\n",
    "\n",
    "- A partir de un RDD de listas/tuplas\n",
    "- A partir de un RDD de objetos Row\n",
    "- A partir de ficheros JSON\n",
    "- A partir de otros almacenamientos (Parquet, Hive,...)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+\n",
      "|edad| nombre|\n",
      "+----+-------+\n",
      "|  17|  Celia|\n",
      "|  53|   Juan|\n",
      "|  39|Manuela|\n",
      "|  17|    Ana|\n",
      "+----+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import SQLContext\n",
    "sqlContext = SQLContext(sc)\n",
    "\n",
    "# A partir de ficheros json\n",
    "dfJson = sqlContext.read.json(\"data/gente.json\")\n",
    "\n",
    "dfJson.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+--------------+----------+\n",
      "|code|          name|phone_code|\n",
      "+----+--------------+----------+\n",
      "|  AF|   Afghanistan|       +93|\n",
      "|  AL|       Albania|      +355|\n",
      "|  DZ|       Algeria|      +213|\n",
      "|  AS|American Samoa|        +1|\n",
      "|  AD|       Andorra|      +376|\n",
      "|  AO|        Angola|      +244|\n",
      "|  AI|      Anguilla|        +1|\n",
      "|  AG|       Antigua|        +1|\n",
      "|  AR|     Argentina|       +54|\n",
      "|  AM|       Armenia|      +374|\n",
      "|  AW|         Aruba|      +297|\n",
      "|  AU|     Australia|       +61|\n",
      "|  AI|       Austria|       +43|\n",
      "|  AZ|    Azerbaijan|      +994|\n",
      "|  BH|       Bahrain|      +973|\n",
      "|  BD|    Bangladesh|      +880|\n",
      "|  BB|      Barbados|        +1|\n",
      "|  BY|       Belarus|      +375|\n",
      "|  BE|       Belgium|       +32|\n",
      "|  BZ|        Belize|      +501|\n",
      "+----+--------------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# A partir de ficheros json\n",
    "dfJson = sqlContext.read.json(\"data/countries.json\")\n",
    "\n",
    "dfJson.show()"
   ]
  },
  {
   "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
}