Parte 2. S04. Operaciones con SparkDataFrames.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<span style='font-size:2.4em'>Procesamiento masivo de datos con SparkR</span>\n",
    "\n",
    "<span style='font-size:1.5em'>VIII Jornadas de usuarios de R. Albacete, Castilla-La Mancha, 17 y 18 de noviembre de 2016</span>\n",
    "\n",
    "Taller impartido por: <span style='font-size:1.2em'>Manuel Jesús Parra Royón</span>\n",
    "\n",
    "\n",
    "![Alt](https://sites.google.com/site/manuparra/home/logoparty.png)\n",
    "\n",
    "<HR>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Operaciones sobre SparkDataFrames\n",
    "\n",
    "![Spark+R](https://sites.google.com/site/manuparra/home/SparkRlogo.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Como siempre para todos nuestros `scripts` con **SparkR**, cargamos la biblioteca, y creamos una nueva sesión de SparkR.\n",
    "\n",
    "En este caso:\n",
    "\n",
    "<span style=\"background-color:red;color:white\">&nbsp; &nbsp; Cuidado con la cantidad de MEMORIA que usamos para esta sección ! &nbsp; &nbsp; </span>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    ".libPaths(c(file.path(Sys.getenv(\"SPARK_HOME\"),\"R/lib/\"),.libPaths()))\n",
    "library(SparkR)\n",
    "sparkR.session(appName=\"EntornoInicio\", master = \"local[*]\", sparkConfig = list(spark.driver.memory = \"1g\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Los ``SparkDataFrames`` soportan un alto número de funciones para hacer un procesado de datos estructurado. \n",
    "\n",
    "Vamos a poner en práctica las más utilizadas. La lista completa de operaciones que se pueden aplicar se puede ver desde API de SparkR en https://spark.apache.org/docs/latest/api/R/index.html\n",
    "\n",
    "![funcSparkR](https://sites.google.com/site/manuparra/home/functionSparkR.jpg)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Operaciones con SparkDataFrames"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Cargamos un conjunto de datos masivo desde el repositirio de datasets.\n",
    "\n",
    "\n",
    "Podemos verlos desde: http://localhost:25980/tree\n",
    "\n",
    "\n",
    "El dataset que vamos a usar para el procesamiento de dato masivos, corresponde con un conjunto de datos de los registros de viaje en *TAXI*, donde se capturan las fechas y horas de recogida y devolución de pasajeros, lugares de recogida y entrega (coordenadas), distancias de viaje, tarifas detalladas, tipos de tarifas, tipos de pago y conteos de pasajeros que van en el taxi.  \n",
    "\n",
    "El dataset tiene **MUCHAS** posibilidades de procesamiento y también extracción de conocimiento.\n",
    "\n",
    "Estos conjuntos de datos adjuntos fueron recopilados y proporcionados por la Comisión de Taxisde Nueva York (TLC) http://www.nyc.gov/html/tlc/html/about/trip_record_data.shtml.\n",
    "\n",
    "\n",
    "![Alt](http://www.nyc.gov/html/tlc/images/features/fi_about_photo_trip_records.png)\n",
    "\n",
    "\n",
    "\n",
    "**Características del conjunto de datos original:**\n",
    "\n",
    "- El conjunto de datos NYCTaxiTrips en total tiene sobre **267GB**, que pueden ser manejados sin problema por SparkR (en un cluster real, no sobre una máquina virtual sencilla). \n",
    "- En total contiene 1100 millones de registros.\n",
    "- Más información de como se gestionan 1100 millones de instancias en la siguiente web y se soluciona este problema  problema real: http://toddwschneider.com/posts/analyzing-1-1-billion-nyc-taxi-and-uber-trips-with-a-vengeance/\n",
    "\n",
    "\n",
    "Más datasets masivos de NYCTaxiTrips en:  http://www.nyc.gov/html/tlc/html/about/trip_record_data.shtml\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Primero revisamos los distintos dataset que se han preparado en: http://localhost:25980/tree/datasets\n",
    "\n",
    "* yellow_tripdata_2016-01.csv\n",
    "* yellow_tripdata_2016-02_small1.csv\n",
    "* yellow_tripdata_2016-02_small2.csv\n",
    "* yellow_tripdata_2016-02_small3.csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Cargamos una versión reducida de los datos en CSV\n",
    "df_nyctrips <- read.df(\"/root/TallerSparkR/datasets/yellow_tripdata_2016-02_small3.csv\", \"csv\", header = \"true\", inferSchema = \"true\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Estudiamos de manera superficial el dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Comprobamos los campos del dataset\n",
    "printSchema(df_nyctrips)\n",
    "\n",
    "# Comprobamos como son los datos:\n",
    "head(df_nyctrips)\n",
    "\n",
    "# Contamos el total del registros:\n",
    "count(df_nyctrips)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Selección de instancias y columnas"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Para la selección de columnas y filas, usamos ``select`` y ``filter``. \n",
    "\n",
    "Todas las operaciones se pueden combinar para producir un nuevo dataset o ``SparkDataFrame``. **Son equivalentes a usar SPARKSQL **.\n",
    "\n",
    "Estas operaciones son esenciales si queremos transformar el dataset en otra versión preprocesada del mismo."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Seleccionamos sólo la columna longitud, por el id de la columna\n",
    "# Por ID de columna \n",
    "head(select(df_nyctrips,df_nyctrips$pickup_longitude))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Seleccionamos sólo la columna longitud, por el nombre de la columna.\n",
    "# Por nombre de columna del dataset\n",
    "head(select(df_nyctrips,\"pickup_longitude\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Para aplicar filtros de para las filas usamos ``filter`` que admite expresiones con operadores condicionales: \n",
    "\n",
    "```\n",
    "    < = > ! & | ...\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Aplicamos un filtro para ver los viajes aquellos viajes de taxi de más de 10 millas.\n",
    "head(filter(df_nyctrips, df_nyctrips$trip_distance > 10 & df_nyctrips$total_amount> 20 ))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Aplicamos un filtro para ver los viajes aquellos viajes de taxi de más de 10 millas y el importe mayor de $ 20\n",
    "head(filter(df_nyctrips, df_nyctrips$trip_distance > 10 & df_nyctrips$total_amount> 20 ))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Aplicamos un filtro para ver el viaje más caro en Taxi que se ha hecho:\n",
    "head( agg(df_nyctrips ,max = max(df_nyctrips$total_amount)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Aplicamos un filtro para ver el viaje menos caro en Taxi que se ha hecho:\n",
    "head(agg(df_nyctrips, min = min(df_nyctrips$total_amount)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Uso de Agrupamiento y Agregación"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Los SparkDataFrames soportan funciones de agregado despues de agrupar. \n",
    "\n",
    "Por ejemplo podemos:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Agrupamos por Vendedor y mostramos el número de viajes.\n",
    "head(summarize(groupBy(df_nyctrips, df_nyctrips$VendorID), count = n(df_nyctrips$VendorID)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Agrupamos por Vendedor y mostramos el número de viajes.\n",
    "head(summarize(groupBy(df_nyctrips, df_nyctrips$VendorID), max = max(df_nyctrips$total_amount)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Agrupamos y ordenamos\n",
    "\n",
    "numsum <- summarize(groupBy(df_nyctrips, df_nyctrips$VendorID), num = n(df_nyctrips$VendorID))\n",
    "head(arrange(numsum,asc(numsum$num)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Agrupamos por numero de pasajeros y mostramos el numero de viajes\n",
    "trips_passenger <- summarize(groupBy(df_nyctrips, df_nyctrips$passenger_count), count = n(df_nyctrips$passenger_count))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Cuidado con el COLLECT !\n",
    "trips_df <- head(collect(trips_passenger))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "head(trips_df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Operaciones con columnas\n",
    "\n",
    "Otras operaciones muy familiares en R, corresponden con la manipulación o transformación de valores en los registros de un dataset. En este caso la manipulación es muy sencilla:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Convertimos la columna de millas a kilómetros, igual que en R.\n",
    "df_nyctrips$trip_distance <- df_nyctrips$trip_distance*1.6"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "head(df_nyctrips)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Añadir columnas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Usamos mutate para añadir columnas que operan con elementos de las demás columnas.\n",
    "\n",
    "# mutate(sql_nyc,  uniform = rand(10),  normal  = randn(27))\n",
    "\n",
    "head(mutate(df_nyctrips,  uniform = rand(10),  normal  = randn(27)))\n",
    "head(mutate(df_nyctrips,  uniform =df_nyctrips$total_amount*1.1355,  normal  = randn(27)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Otro modo de hacerlo es:\n",
    "\n",
    "head(withColumn(df_nyctrips,\"uniform\",rand(20)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### dapply -- dapplayCollect\n",
    "\n",
    "Aplicar una función a un conjunto datos masivo con ``dapply`` y ``dapplyCollect`` \n",
    "\n",
    "**dapply**\n",
    "\n",
    "Aplica una función a cada partición de un ``SparkDataFrame``. La función que será aplicada para cada partición y debería tener sólo un parámetro. La salida de la función deberá ser igualmente un data.frame. Además hay que especificar el ``schema`` del formato de los datos del ``SparkDataFrame`` resultante y deberá corresponder con tipo de datos del valor devuelto.\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Hacemos una copia del SparkDataFrame para usarla en una vista temporal en SQL\n",
    "createOrReplaceTempView(df_nyctrips,\"slqdf_filtered_nyc\")\n",
    "\n",
    "# Hacemos una selección de los registros, donde calculamos el tiempo del viaje de cada viaje\n",
    "sql_nyc <- sql(\"select VendorID,INT(unix_timestamp(tpep_dropoff_datetime)- unix_timestamp(tpep_pickup_datetime)) AS trip_time,passenger_count,trip_distance,total_amount from slqdf_filtered_nyc\")\n",
    "\n",
    "# Mostramos un trozo de SparkDataFrame\n",
    "head(sql_nyc)\n",
    "\n",
    "schema(sql_nyc)\n",
    "\n",
    "# Indicamos el Schema, que debe coincidir con lo que queremos\n",
    "schema <- structType(\n",
    "    structField(\"VendorID\", \"integer\"),\n",
    "    structField(\"trip_time\", \"integer\"), \n",
    "    structField(\"passenger_count\", \"integer\"),\n",
    "    structField(\"trip_distance\", \"double\"),\n",
    "    structField(\"total_amount\", \"double\"),\n",
    "    structField(\"total_amount_euro\", \"double\")\n",
    ")\n",
    "\n",
    "# Creamos la función que hará los cambios.\n",
    "new_sql_nyc <- dapply(\n",
    "    sql_nyc, \n",
    "    function(x) { \n",
    "        x <- cbind(x, x$total_amount*1.1355) \n",
    "    }, \n",
    "    schema)\n",
    "\n",
    "# Vemos el cambio\n",
    "head(new_sql_nyc)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "### gapply -- gapplyCollect\n",
    "\n",
    "\n",
    "Aplica una función a cada uno de los grupos de un ``SparkDataFrame``. La función será aplicada a cada grupo del ``SparkDataFrame`` y debería tener sólo dos parámetros: agrupamiento por llave y data.frame al que corresponde esa llave. La salida de la función debería ser un data.frame. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Esquema del SparkDataFrame\n",
    "schema <- structType(\n",
    "    structField(\"VendorID\", \"integer\"),\n",
    "    structField(\"trip_time\", \"integer\"), \n",
    "    structField(\"passenger_count\", \"integer\"),\n",
    "    structField(\"trip_distance\", \"double\"),\n",
    "    structField(\"total_amount\", \"double\"),\n",
    "    structField(\"max_amount\", \"double\")\n",
    ")\n",
    "\n",
    "# Aplicamos la función gapply. Calculamos el máximo de cada Vendedor.\n",
    "result <- gapply(\n",
    "    sql_nyc,\n",
    "    \"VendorID\",\n",
    "    function(key, x) {\n",
    "        y <- data.frame(key, max(x$total_amount))\n",
    "    },\n",
    "    schema)\n",
    "\n",
    "# Mostramos el resultado.\n",
    "head(result[order(result$trip_distance, decreasing = TRUE), ])\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "head(sql_nyc)\n",
    "\n",
    "# Ahora probamos el gapplycollect: \n",
    "# Como el gapply, aplica una funcion a cada partición y luego hace un collect del resultado en un data.frame en R.\n",
    "result <- gapplyCollect(\n",
    "            \n",
    "    sql_nyc,\n",
    "    \"VendorID\",\n",
    "    function(key, x) {\n",
    "        y <- data.frame(key, max(x$trip_distance))\n",
    "        colnames(y) <- c(\"VendorID\", \"max_trip_distance\")\n",
    "        y\n",
    "    })\n",
    "\n",
    "# Vemos el resultado.\n",
    "head(result[order(result$trip_distance, decreasing = TRUE), ])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Operando con SparkSQL sobre cojuntos masivos de datos."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Todas las funciones de manejo de datos que se han usado con SparkR, pueden hacerse de una forma sencilla e intuitiva  con SparkSQL"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# sql_nyc es nuestro DataFrameSpark de SQL\n",
    "createOrReplaceTempView(sql_nyc,\"slqdf_filtered_nyc\")\n",
    "\n",
    "# Hacemos una consulta para extraer el viaje de mayor distancia de cada venderor.\n",
    "results <- sql(\"select VendorID, MAX(trip_distance) from slqdf_filtered_nyc GROUP BY VendorID \")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Vemos el resultado.\n",
    "head(results)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Buscamos el total de kilómetros recorridos por cada vendedor:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "results <- sql(\"select VendorID, SUM(trip_distance) from slqdf_filtered_nyc GROUP BY VendorID \")\n",
    "\n",
    "# Vemos el resultado\n",
    "head(results)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Calculamos el tiempo en segundos"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "results <- sql(\"select VendorID, SUM(trip_time) from slqdf_filtered_nyc GROUP BY VendorID \")\n",
    "\n",
    "# Vemos los resultados\n",
    "head(results)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Calculamos el tiempo en minutos"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "results <- sql(\"select VendorID, SUM(trip_time)/60.0 as min_trip from slqdf_filtered_nyc GROUP BY VendorID \")\n",
    "\n",
    "# Vemos los resultados\n",
    "head(results)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Buscamos la ganacia total cada vendedor:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "results <- sql(\"select VendorID, SUM(total_amount)*1.10373 as Total_Amount_Euro from slqdf_filtered_nyc GROUP BY VendorID \")\n",
    "\n",
    "# Vemos el resultado\n",
    "head(results)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Calculamos la media y la desviación típica del tiempo de recorrido y ganancia por numero de personas: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "results <- sql(\"select passenger_count, AVG(trip_time), AVG(total_amount) ,AVG(trip_distance)   \n",
    "                from slqdf_filtered_nyc \n",
    "                GROUP BY passenger_count \n",
    "                order by passenger_count ASC \")\n",
    "head(results)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Coeficiente de correlación"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "results <- sql(\"select corr(total_amount,trip_distance) as correlation_coef\n",
    "                from \n",
    "                slqdf_filtered_nyc\")\n",
    "# Ver resultados\n",
    "head(results)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "results <- sql(\"select corr(total_amount,trip_time) as correlation_coef\n",
    "                from \n",
    "                slqdf_filtered_nyc\")\n",
    "head(results)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "results <- sql(\"select corr(trip_time,trip_distance) as correlation_coef\n",
    "                from \n",
    "                slqdf_filtered_nyc\")\n",
    "head(results)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**¿ Qué deducimos de estos coeficiente de corelación ?**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "## Uso de magittr para el trabajo con los datos"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "El paquete magrittr permite: \n",
    "\n",
    "* mejorar el tiempo de desarrollo y \n",
    "* mejorar enormemente la legibilidad y mantenibilidad del código. \n",
    "\n",
    "Para usarlo hay que importar la biblioteca magrittr dentro del proyecto y apartir de ese momentos podemos utilizar el operador \n",
    "\n",
    "```\n",
    "%>%\n",
    "``` \n",
    "\n",
    "para concaternar operaciones y poder trabajar con flujos de datos y pipelines.\n",
    "\n",
    "Provee de un operador que sirve para hacer `pipes` con el cual se puede `encauzar` un valor hacia adelante dentro de una expresión o llamada a función.\n",
    "\n",
    "Veamos todas las operaciones que hemos realizado sobre los datos y su equivalente con `pipes`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Hacemos una copia del SparkDataFrame para usarla en una vista temporal en SQL\n",
    "createOrReplaceTempView(df_nyctrips,\"slqdf_filtered_nyc\")\n",
    "\n",
    "# Hacemos una selección de los registros, donde calculamos el tiempo del viaje de cada viaje\n",
    "sql_nyc <- sql(\"select VendorID,INT(unix_timestamp(tpep_dropoff_datetime)- unix_timestamp(tpep_pickup_datetime)) AS trip_time,passenger_count,trip_distance,total_amount from slqdf_filtered_nyc\")\n",
    "\n",
    "head(sql_nyc)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Usamos magrittr\n",
    "library(magrittr)\n",
    "\n",
    "# results <- sql(\"select VendorID, MAX(trip_distance) from slqdf_filtered_nyc GROUP BY VendorID \")\n",
    "#summarize(groupBy(df_nyctrips, df_nyctrips$passenger_count), count = n(df_nyctrips$passenger_count))\n",
    "\n",
    "df_nyctrips %>% \n",
    "        groupBy( df_nyctrips$passenger_count) %>%\n",
    "        summarize(count = n(df_nyctrips$passenger_count)) %>%\n",
    "        head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df_nyctrips %>% \n",
    "        groupBy( df_nyctrips$passenger_count) %>%\n",
    "        summarize( avg_total_amount=avg(df_nyctrips$total_amount) ,avg_trip_distance=avg(df_nyctrips$trip_distance)) %>%\n",
    "        head()\n",
    "        "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df_nyctrips %>% \n",
    "         groupBy( df_nyctrips$passenger_count) %>%\n",
    "        summarize(min = min(df_nyctrips$trip_distance),max = max(df_nyctrips$trip_distance)) %>%\n",
    "        head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df_nyctrips %>% \n",
    "         groupBy( df_nyctrips$passenger_count, hour(df_nyctrips$tpep_pickup_datetime)) %>%\n",
    "        summarize(total_pickup = n(df_nyctrips$tpep_pickup_datetime)) %>%\n",
    "        head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "count(sql_nyc)\n",
    "num_regs <- as.integer(count(sql_nyc))\n",
    "\n",
    "# Mostramos el número de registros\n",
    "print(num_regs)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "** ¿Qué es lo mejor: `pipes`, SPARKSQL o funciones? **"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "<HR>\n",
    "# Zona de pruebas del NOTEBOOK en SparkR\n",
    "![FooterSparkR](https://sites.google.com/site/manuparra/home/footer_SparkR_v2.png)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "Escribe todas las pruebas en R que necesites a partir de aquí\n",
    "\n",
    "<HR>"
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "R",
   "language": "R",
   "name": "ir"
  },
  "language_info": {
   "codemirror_mode": "r",
   "file_extension": ".r",
   "mimetype": "text/x-r-source",
   "name": "R",
   "pygments_lexer": "r",
   "version": "3.3.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}