pyspark-introduction.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Conceptos Basicos de Pyspark\n",
    "\n",
    "Este notebook se ha utilizado para dar una introducción de la sintaxis de __PySpark__.\n",
    "\n",
    "__Author__:  \n",
    "Carlos Sevilla  \n",
    "c.sevilla.barcelo@gmail.com  \n",
    "\n",
    "## Spark Session\n",
    "\n",
    "La session de spark es el punto de entrada a la aplicación de Spark. Es obligatoria para usar el entorno de Spark. Con ella, podemos leer archivos csv, leer desde bases de datos, de sistemas streaming, etc. \n",
    "\n",
    "Una vez que hemos instalado pyspark, hay varias formas de lanzarlo.\n",
    "\n",
    "__Forma 1:__  \n",
    "La forma tradicional, la típica, abrir una consola de python3 y crear la spark session\n",
    "\n",
    "Abrimos una consola de python3\n",
    "```bash\n",
    "user@computer: python3\n",
    "```\n",
    "Creamos una spark session \n",
    "```python\n",
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder.appName('nombre_session_spark').getOrCreate()\n",
    "```\n",
    "\n",
    "__Forma 2:__  \n",
    "Lanzamos en la consola:\n",
    "```bash\n",
    "pyspark\n",
    "```\n",
    "Esto te da una session de spark ya inicializada en la variable `spark`, con lo que podemos empezar a realizar operaciones desde la primera linea\n",
    "\n",
    "\n",
    "__Para los notebooks, solo se puede usar la forma 1.__ Basicamente, porque jupyter notebook se lanza con el comando `jupyter notebook` \n",
    "\n",
    "__ToDo:__ Crea una sesion de spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Leer un csv"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Una vez que tenemos nuestra `spark session`, podemos llamar a su clase `read`, la cual nos permite crear dataframe desde archivos. El metodo `.csv`, evidentemente, nos permite crear un Dataframe a partir de archivo/archivos `.csv` \n",
    "\n",
    "```python\n",
    "df = spark.read.csv('ruta/a/mi/archivo')\n",
    "```\n",
    "\n",
    "Este lector tiene algunos parametros importantes:\n",
    "\n",
    "* sep - Separador en el csv. Default = ','\n",
    "* schema - Esquema con el tipo de los datos. Default = None\n",
    "* inferSchema - Crea el schema de forma automatica. default = False\n",
    "* header - Usa la primera linea como nombre de las columnas. Default = False\n",
    "\n",
    "Podeis leer mas del lector de csv y sus parametros [aqui](http://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.streaming.DataStreamReader.csv)\n",
    "\n",
    "__ToDo:__ Leer `users.csv` de la carpeta `data`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Operaciones "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Ver el contenido del DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Resumen del DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.describe().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Columnas del DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Ver schema del DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Crear Columnas\n",
    "\n",
    "Para crear una columna, usamos el metodo `.withColumn()`, el cual recibe por parametros el __nombre de la columna__ como parametro 1, y la operacion/dato como parametro 2.\n",
    "\n",
    "Un ejemplo: \n",
    "```python\n",
    "df.withColumn('altura_cuadrada', df.altura_cuadrada * df.altura_cuadrada)\n",
    "```\n",
    "Esto devuelve el DataFrame `df` que teniamos, junto con esta columna nueva. No se guarda en el DataFrame de forma automática, por lo que para guardarlo, tenemos que guardar este dataframe que nos devuelve en una variable\n",
    "\n",
    "```python\n",
    "df = df.withColumn('altura_cuadrada', df.altura_cuadrada * df.altura_cuadrada)\n",
    "```\n",
    "\n",
    "Para modificar el contenido de una columna, también se usa el metodo `.withColumn()`. Como nombre de la columna, hay que pararle el nombre de la columna que queremos modificar.\n",
    "\n",
    "__ToDo:__ Crea una columna nueva con el IMC de cada usuario"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Funciones de PySpark SQL\n",
    "\n",
    "PySpark incluye en su API bastantes funciones. Muchas de ellas, son funciones que podemos encontrar en __SQL__, algunas otras son alias de otras funciones, o simplemente snippets para agilizar la programación\n",
    "\n",
    "Puedes encontrarlas todas [aquí](http://spark.apache.org/docs/latest/api/python/pyspark.sql.html#module-pyspark.sql.functions) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark.sql.functions  as F"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### select\n",
    "\n",
    "Podemos seleccionar las columnas que queremos ver, o realizar operaciones que no se guarden en el DataFrame. Es el mismo concepto que el SELECT de __SQL__"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.select(['nombre','edad','IMC']).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.select(df.edad < 30).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### filter\n",
    "Es el metodo que nos permite filtrar un DataFrame en base a unas condiciones. `filter()` nos devuelve el DataFrame original filtrado."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "df.filter(df.edad < 30).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.filter((df.edad < 30) & (df.altura > 1.5)).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Si queremos guardar este DataFrame filtrado, basta con guardarlo en una variable"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_filtered = df.filter(df.edad < 30)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "__ToDo:__ Prueba a hacer algún filtro"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### groupBy\n",
    "\n",
    "Nos permite agrupar el DataFrame en base a una o varias columnas. Necesitamos definir el comportamiento de los valores de las columnas agrupadas. Existen diferentes métodos:\n",
    "- count()\n",
    "- sum()\n",
    "- min()\n",
    "- max()\n",
    "- mean()\n",
    "\n",
    "También podemos definir una función agregada con `.agg()` para aplicar diferentes comportamientos sobre las agrupaciones. La función `.agg()`  recibe un diccionario como parametro.  \n",
    "En este diccionario, la clave es el nómbre de la columna, y el valor es la operación a realizar con esa columna"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.groupBy('apellidos').count().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### join\n",
    "\n",
    "tipos:\n",
    "- inner\n",
    "- left\n",
    "- left_anti\n",
    "- outer\n",
    "\n",
    "```python\n",
    "df.join(d2,['columnas','join'],'inner')\n",
    "```\n",
    "\n",
    "__ToDo:__ Carga el dataset `family_info.csv` y juntalo junto a dataframe original, sin perder información original del dataframe original.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Max / Min / Mean"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.select(F.max(F.col('edad'))).show()\n",
    "df.select(F.min(F.col('edad'))).show()\n",
    "df.select(F.mean(F.col('edad'))).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### sqrt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.select(F.sqrt(F.col('IMC'))).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### lit\n",
    "\n",
    "Devuelve una columna con un valor estatico en todas las filas.   \n",
    "Ejemplo: Meter el contenido de una variable en el DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.withColumn('continente',F.lit('Westeros'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### UDF\n",
    "User Defined Functions.\n",
    "\n",
    "Aplicamos funciones en Spark que hemos definido nosotros. \n",
    "Para ello, hay que usar la funcion `udf` de `pyspark.sql.functions`, que tiene la siguiente estructura:\n",
    "```python\n",
    "udf(lambda x: funcion_definida(x), tipo_datos())\n",
    "```\n",
    "`tipo_datos()` es el tipo de datos que devuelve esta funcion. Puede ser un String, un Double, una Lista, etc. Consulta los tipos [aquí](http://spark.apache.org/docs/latest/api/python/pyspark.sql.html#module-pyspark.sql.types)\n",
    "\n",
    "Ejemplo: \n",
    "\n",
    "```python\n",
    "from pyspark.sql.types import StringType\n",
    "\n",
    "def altera_apellido(x):\n",
    "    longitud = len(x)\n",
    "    mitad = int(longitud/2)\n",
    "    parte_1 = x[:longitud]\n",
    "    parte_2 = x[longitud:]\n",
    "    nuevo_string = parte_2 + parte_1\n",
    "    return nuevo_string\n",
    "\n",
    "\n",
    "altera_apellido_udf = udf(lambda x: altera_apellido(x),StringType())\n",
    "\n",
    "```\n",
    "\n",
    "Una vez que tenemos la funcion definida, podemos usarla en spark.\n",
    "\n",
    "```python\n",
    "df.withColumn('apellido_alterado',altera_apellido_udf)\n",
    "\n",
    "```\n",
    "\n",
    "__ToDo:__ Prueba a crear una función y usarla en tu dataframe! "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### isNull \n",
    "\n",
    "En spark, existen simultaneamente el tipo `NaN` (not a number), `null` y `None`. Pese a que `null` es como un \"alias\" de `None`, `NaN` es tratado como un `Double`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.filter(df.edad.isNull()).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Window\n",
    "\n",
    "Uno de los puntos fuertes de Spark son las operaciones sober una ventana (Window) de valores. Sirve para realizar acciones simultaneas entre diferentes valores de una misma columna. \n",
    "\n",
    "Ejemplo: Queremos añadirle un campo id a los usuarios. Queremos un id unico para usuarios del mismo apellido.\n",
    "\n",
    "Para ello, podemos usar `Window`, aplicando una ventana en el campo `apellido`. Esto hará que la operación que vayas a realizar, se haga por cada valor diferente en el campo `apellidos`.  \n",
    "Para entendernos, lo que hace __Spark__ es coger el dataframe original y \"crear\" X dataframes mas pequeños, siendo X el __número de valores únicos__ en el dataframe. Si tenemos 4 valores únicos, va a crear 4 dataframes.  \n",
    "En estos 4 dataframes, aplica la acción, y despues vuelve a unir estos 4 dataframes se vuelven a juntar en 1 solo como en dataframe original\n",
    "\n",
    "Para crear una ventana, necesitamos indicar el campo por el que se va a particionar el dataframe. Luego, en el dataframe, en la accion que vamos a realizar añadimos el metodo `.over()`, el cual recibe por parametro la ventana que hemos creado. \n",
    "\n",
    "```python\n",
    "from pyspark.sql import Window\n",
    "\n",
    "ventana = Window.partitionBy('apellidos')\n",
    "\n",
    "df = df.withColumn('id',F.row_number().over(ventana))\n",
    "```\n",
    "\n",
    "El metodo `row_number` es un metodo de spark que devuelve el numero de la fila en el dataframe\n",
    "\n",
    "Estas ventanas, pueden particionar por mas de un campo, y pueden ser ordenados.\n",
    "\n",
    "```python\n",
    "ventana = Window.partitionBy('apellidos').orderBy('edad','peso')\n",
    "```\n",
    "\n",
    "__ToDo:__ Prueba a realizar una operación con una ventana!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Operaciones sobre fechas\n",
    "\n",
    "__Todas estas operaciones se aplican sobre toda la columna__\n",
    "\n",
    "### dayofweek\n",
    "\n",
    "Te devuelve el día de las fechas en la columna que recibe como parametro\n",
    "\n",
    "```python\n",
    "df.withColumn('dia_semana',F.dayofweek(F.col('init_date'))\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### month\n",
    "\n",
    "Devuelve el mes de las fechas en la columna que recibe como parametro\n",
    "\n",
    "```python\n",
    "df.withColumn('mes_batalla',F.month(F.col('init_date'))\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### months_between\n",
    "Devuelve el numero de meses entre 2 columna de fechas\n",
    "\n",
    "```python\n",
    "df.withColumn('duracion_batallas',F.months_between(F.col('init_date'),F.col('finish_date'))\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### date_add\n",
    "\n",
    "Recibe como parametro el número de días que quieres desplazar la fecha hacia el futuro, y devuelve la fecha ya desplazada\n",
    "\n",
    "```python\n",
    "df.withColumn('fin_del_luto',date_add(df.finish_date, 5))\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Otros\n",
    "\n",
    "### Spark DataFrame -> Pandas DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_p = df.toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Practica Final"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "__ToDo:__\n",
    "- Carga `battles.csv`\n",
    "- Crea una columna que contenga el día de la semana (en texto) del día que empezó cada batalla\n",
    "- Crea una columna con la diferencia de meses entre una batalla y otra (usar la funcion `lag` de `pyspark.sql.functions`\n",
    "- Extrae del dataset la batalla mas larga"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "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.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}