3.2-Intro to Spark - Ejemplos - soluciones.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Intro to Spark\n",
    "Ejemplos prácticos\n",
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Crear una sesión de Spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### Con SparkSession\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "# crear sesión\n",
    "\n",
    "spark = SparkSession \\\n",
    "        .builder \\\n",
    "        .appName(\"Python Spark SQL basic example\") \\\n",
    "        .config(\"local[*]\") \\\n",
    "        .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## parar la sesión\n",
    "spark.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Con SparkContext y SQLContext\n",
    "from pyspark import SparkContext\n",
    "from pyspark import SQLContext\n",
    "\n",
    "## crear el contexto\n",
    "sc = SparkContext(\"local[*]\", \"pyspark_df\")\n",
    "ss = SQLContext(sc)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### parar el contexto\n",
    "sc.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### Crear un Dataframe a partir de un csv\n",
    "dfs_flights = spark.read.option(\"inferSchema\", \"true\").option(\"header\", \"true\").csv('data/2015-summary.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### Inspeccionar el Dataframe\n",
    "dfs_flights.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Renombrar la cplumna count\n",
    "\n",
    "dfs_flights = dfs_flights.withColumnRenamed('count', 'NumFlights')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dfs_flights.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### Ordenar según la columna count\n",
    "dfs_fligths_sorted = dfs_flights.sort(\"NumFlights\", ascending=False) # no tarda nada porque es una transformación"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dfs_fligths_sorted.show(10) ## al hacerle el show realiza el sort y nos devuelve los 10 primeros"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### group by\n",
    "count_per_dest = dfs_flights.groupBy(\"DEST_COUNTRY_NAME\").sum()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### Enseña el dataframe count_per_dest\n",
    "count_per_dest.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Renombra la columna sum(NumFlights)\n",
    "count_per_dest = count_per_dest.withColumnRenamed('sum(NumFlights)', 'total_flights')\n",
    "count_per_dest.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### Ordena por número de vuelos\n",
    "most_flights = count_per_dest.sort('total_flights', ascending=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### Top 5\n",
    "most_flights.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Querying the data with SQL"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create temp table\n",
    "\n",
    "df_flights.createOrReplaceTempView(\"flights_2015\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "query = \"\"\"\n",
    "SELECT *\n",
    "FROM flights_2015\n",
    "ORDER BY count DESC\n",
    "LIMIT 10\n",
    "\"\"\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "query_result = spark.sql(query)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "type(query_result)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "query_result.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Calcula el total de vuelos en función del destino y saca el destino más frecuente"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Funcionalidad dataframes\n",
    "\n",
    "Muchas veces nos encontraremos con que queremos utilizar algua funcion integrada en los dataframes, cómo en pandas.\n",
    "\n",
    "Para poder acceder a estas funciones, tendremos que importarlas de la librería"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import max"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.sql(\"SELECT max(count) from flights_2015\").take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dfs_flights.select(max(\"NumFlights\")).take(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Intefración con pandas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = dfs_flights.toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = dfs_flights.groupBy(\"DEST_COUNTRY_NAME\").sum(\"NumFlights\")\\\n",
    "           .withColumnRenamed(\"sum(NumFlights)\", \"destination_total\")\\\n",
    "           .sort(desc(\"destination_total\"))\\\n",
    "           .limit(10)\\\n",
    "           .toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%matplotlib inline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.set_index('DEST_COUNTRY_NAME').plot(kind='barh')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## from pandas to SparkDataframe\n",
    "\n",
    "df_test = spark.createDataFrame(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Ejercicio\n",
    "---\n",
    "\n",
    "Haz un análisis exploratorio del fichero 'data/201508_trip_data.csv' usando spark y plotea los resultados usando las funcionalidades de python"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## importa lo necesario para crear el entorno de spark\n",
    "\n",
    "### Con SparkSession\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "# crear sesión\n",
    "\n",
    "spark = SparkSession.builder.appName(\"Bikes\").config(\"local[*]\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## carga el fichero 'data/201508_trip_data.csv'\n",
    "path = 'data/201508_trip_data.csv'\n",
    "\n",
    "dfs_bikes = spark.read.option('inferSchema', True).option('header', True).csv(path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Imprime el esquema y las columnas del dfs_bikes\n",
    "\n",
    "dfs_bikes.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Haz una descripción de la columnas Duration del dfs_bikes\n",
    "## En que unidades está la duración ?\n",
    "dfs_bikes.select('Duration').describe().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Cual son las 5 estaciones más transitada?\n",
    "\n",
    "most_frequent = dfs_bikes.select('Zip Code', 'Duration').groupby('Zip Code').count()\\\n",
    "                .withColumnRenamed('count', 'num_trips')\\\n",
    "                .sort('num_trips', ascending=False).limit(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Repite el ejercicio anterior usando SQL\n",
    "\n",
    "## Cambia el nombre a la columna para que no tenga espacios\n",
    "dfs_bikes = dfs_bikes.withColumnRenamed('Zip Code', 'zipcode')\n",
    "\n",
    "## Crea una tabla temporal\n",
    "dfs_bikes.createOrReplaceTempView('bike_trips')\n",
    "\n",
    "## Escribe una query sql que realice la tarea de antes\n",
    "query = \"\"\"\n",
    "\n",
    "SELECT zipcode, count(1) as num_trips\n",
    "FROM bike_trips\n",
    "GROUP BY zipcode\n",
    "ORDER BY num_trips DESC\n",
    "LIMIT 5\n",
    "\"\"\"\n",
    "\n",
    "most_frequent_sql = spark.sql(query)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### Transform it to pandas and plot the result\n"
   ]
  },
  {
   "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.6.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}