sparkassignments.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "# Data processing course assignments\n",
    "---"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import pyspark\n",
    "from pyspark import SparkConf, SparkContext\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql import SQLContext\n",
    "from pyspark.sql.functions import asc, lit, udf, sum\n",
    "from pyspark.sql.types import *\n",
    "import re"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "conf = SparkConf().setAppName('Assignment').setMaster('local')\n",
    "spark = SparkContext(conf=conf)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Ejercicio 1.\n",
    "#### Leer el archivo data/containers.csv y contar el número de líneas.\n",
    "---"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "path = \"/home/jovyan/work/data_processing_course-master/assignments/data/containers.csv\"\n",
    "csvdata = spark.textFile(path) #.option(\"header\", \"false\").load(path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['ship_imo;ship_name;country;departure;container_id;container_type;container_group;net_weight;gross_weight;owner;declared;contact;customs_ok',\n",
       " 'AMC1861710;Jayden;BD;201602183;FCUK1755843;4960;28VH;44804866.62;2240243.33;Streich-Wilkinson;Music, Tools, Automotive & Health;octavia@stammbednar.name;true']"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "csvdata.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "614"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "csvdata.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Ejercicio 2.\n",
    "#### Leer el archivo data/containers.csv y filtrar aquellos contenedores cuyo ship_imo es DEJ1128330 y el grupo del contenedor es 22P1. Guardar los resultados en un archivo de texto en resultados/resutado_2.\n",
    "---"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'pyspark.rdd.RDD'>\n",
      "<class 'pyspark.rdd.RDD'>\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[['ship_imo',\n",
       "  'ship_name',\n",
       "  'country',\n",
       "  'departure',\n",
       "  'container_id',\n",
       "  'container_type',\n",
       "  'container_group',\n",
       "  'net_weight',\n",
       "  'gross_weight',\n",
       "  'owner',\n",
       "  'declared',\n",
       "  'contact',\n",
       "  'customs_ok'],\n",
       " ['AMC1861710',\n",
       "  'Jayden',\n",
       "  'BD',\n",
       "  '201602183',\n",
       "  'FCUK1755843',\n",
       "  '4960',\n",
       "  '28VH',\n",
       "  '44804866.62',\n",
       "  '2240243.33',\n",
       "  'Streich-Wilkinson',\n",
       "  'Music, Tools, Automotive & Health',\n",
       "  'octavia@stammbednar.name',\n",
       "  'true']]"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "print(type(csvdata))\n",
    "datardd = csvdata\n",
    "print(type(datardd))\n",
    "\n",
    "datasplit = datardd.map(lambda row: row.split(\";\"))\n",
    "\n",
    "datasplit.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['ship_imo',\n",
       " 'AMC1861710',\n",
       " 'POG1615575',\n",
       " 'SQH1155999',\n",
       " 'JCI1797526',\n",
       " 'MBV1836745',\n",
       " 'GYR1192020',\n",
       " 'GLV1922612',\n",
       " 'NLH1771681',\n",
       " 'FUS1202266']"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "datasplit.map(lambda row: row[0]).take(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "datasplit.filter(lambda row: row[6] == '22P1' and row[0] == 'DEJ1128330').count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[['DEJ1128330',\n",
       "  'Tiara',\n",
       "  'GP',\n",
       "  '2016021818',\n",
       "  'GYFD1228113',\n",
       "  '20PF',\n",
       "  '22P1',\n",
       "  '51503716.88',\n",
       "  '5150371.69',\n",
       "  'Armstrong-Goldner',\n",
       "  'Automotive, Sports, Games & Clothing',\n",
       "  'caria@cronin.io',\n",
       "  'true'],\n",
       " ['DEJ1128330',\n",
       "  'Tiara',\n",
       "  'GP',\n",
       "  '2016021818',\n",
       "  'MBPF1909627',\n",
       "  '24H2',\n",
       "  '22P1',\n",
       "  '37266600.88',\n",
       "  '1863330.04',\n",
       "  'Lehner-Hamill',\n",
       "  'Jewelery, Automotive, Games & Electronics',\n",
       "  'phoebe@volkman.net',\n",
       "  'true']]"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "datasplit.filter(lambda row: row[6] == '22P1' and row[0] == 'DEJ1128330').take(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Guardando los datos como archivo de texto"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "x1 = datasplit.filter(lambda row: row[6] == '22P1' and row[0] == 'DEJ1128330')\n",
    "x1.saveAsTextFile(\"/home/jovyan/work/data_processing_course-master/assignments/data/111.csv\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Ejercicio 3.\n",
    "#### Leer el archivo data/containers.csv y convertir a formato Parquet. Recuerda que puedes hacer uso de la funcion parse_container en helpers.py tal y como vimos en clase. Guarda los resultados en resultados/resultado_3.\n",
    "---"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "path = \"/home/jovyan/work/data_processing_course-master/assignments/data/containers.csv\" #Usamos el mismo path creado previamente"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "sqlContext = SQLContext(spark)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "df = sqlContext.read.format('com.databricks.spark.csv').options(header = \"true\", sep = ';').load(path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(ship_imo='AMC1861710', ship_name='Jayden', country='BD', departure='201602183', container_id='FCUK1755843', container_type='4960', container_group='28VH', net_weight='44804866.62', gross_weight='2240243.33', owner='Streich-Wilkinson', declared='Music, Tools, Automotive & Health', contact='octavia@stammbednar.name', customs_ok='true')"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.first()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "df.write.parquet(\"ejercicios33.parquet\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Ejercicio 4.\n",
    "#### Lee el archivo de Parquet guardado en el ejercicio 3 y filtra los barcos que tienen al menos un contenedor donde la columna customs_ok es igual a false. Extrae una lista con los identificadores de barco, ship_imo, sin duplicados y ordenados alfabéticamente, en formato json.\n",
    "---"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "data4 = sqlContext.read.parquet(\"ejercicios3.parquet\")\n",
    "data4.registerTempTable(\"parquetFile\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|  ship_imo|\n",
      "+----------+\n",
      "|KSP1096387|\n",
      "|GYR1192020|\n",
      "|JET1053895|\n",
      "|SQH1155999|\n",
      "|NLH1771681|\n",
      "|JCI1797526|\n",
      "|GEU1548633|\n",
      "|AEY1108363|\n",
      "|IWE1254579|\n",
      "|AMC1861710|\n",
      "|POG1615575|\n",
      "|MBV1836745|\n",
      "|GLV1922612|\n",
      "|YZX1455509|\n",
      "|TCU1641123|\n",
      "|JMP1637582|\n",
      "|DEJ1128330|\n",
      "|RYP1117603|\n",
      "|FUS1202266|\n",
      "|NCZ1777367|\n",
      "+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sqlContext.sql(\"SELECT DISTINCT ship_imo FROM parquetFile WHERE customs_ok = 'false'\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "data4 = sqlContext.sql(\"SELECT DISTINCT ship_imo FROM parquetFile WHERE customs_ok = 'false'\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "20"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sqlContext.sql(\"SELECT DISTINCT ship_imo FROM parquetFile WHERE customs_ok = 'false'\").count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(ship_imo='AEY1108363'),\n",
       " Row(ship_imo='AMC1861710'),\n",
       " Row(ship_imo='DEJ1128330'),\n",
       " Row(ship_imo='FUS1202266'),\n",
       " Row(ship_imo='GEU1548633'),\n",
       " Row(ship_imo='GLV1922612'),\n",
       " Row(ship_imo='GYR1192020'),\n",
       " Row(ship_imo='IWE1254579'),\n",
       " Row(ship_imo='JCI1797526'),\n",
       " Row(ship_imo='JET1053895'),\n",
       " Row(ship_imo='JMP1637582'),\n",
       " Row(ship_imo='KSP1096387'),\n",
       " Row(ship_imo='MBV1836745'),\n",
       " Row(ship_imo='NCZ1777367'),\n",
       " Row(ship_imo='NLH1771681'),\n",
       " Row(ship_imo='POG1615575'),\n",
       " Row(ship_imo='RYP1117603'),\n",
       " Row(ship_imo='SQH1155999'),\n",
       " Row(ship_imo='TCU1641123'),\n",
       " Row(ship_imo='YZX1455509')]"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "data4.sort(asc(\"ship_imo\")).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[61] at toJavaRDD at NativeMethodAccessorImpl.java:0"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "data4.sort(asc(\"ship_imo\")).toJSON(\"data_processing_course-master/assignments/resultados/resultado_4\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Ejercicio 5.\n",
    "#### Crea una UDF para validar el código de identificación del contenedor container_id. Para simplificar la validación, daremos como válidos aquellos códigos compuestos de 3 letras para el propietario, 1 letra para la categoría, 6 números y 1 dígito de control. Devuelve un DataFrame con los campos: ship_imo, container_id, propietario, categoria, numero_serie y digito_control.\n",
    "---"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+\n",
      "|container_id|\n",
      "+------------+\n",
      "| DXTQ1407119|\n",
      "| VXNB1938296|\n",
      "| OVMU1118217|\n",
      "| EAQO1539643|\n",
      "| SGQH1799946|\n",
      "| SWXT1708984|\n",
      "| EDBR1562470|\n",
      "| FCMB1487245|\n",
      "| UBRI1681197|\n",
      "| LFTG1322014|\n",
      "| RAUX1713695|\n",
      "| QKOJ1756061|\n",
      "| JVFH1614514|\n",
      "| KETX1362337|\n",
      "| KUOG1927848|\n",
      "| GSVC1467358|\n",
      "| JXRI1226202|\n",
      "| QZKL1853985|\n",
      "| HDGM1122708|\n",
      "| ZBJH1066313|\n",
      "+------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sqlContext.sql(\"SELECT DISTINCT container_id FROM parquetFile \").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Pruebas con el módulo de python 're'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[Row(container_id='DXTQ1407119'), Row(container_id='VXNB1938296')]\n"
     ]
    }
   ],
   "source": [
    "test5 = sqlContext.sql(\"SELECT DISTINCT container_id FROM parquetFile \").take(2)\n",
    "type(test5)\n",
    "type(test5[0])\n",
    "print(str(test5))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "containeridre = re.search(r'\\w\\w\\w\\w\\d\\d\\d\\d\\d\\d\\d', str(test5[0]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "ez pz\n"
     ]
    }
   ],
   "source": [
    "if containeridre:\n",
    "    print('ez pz')\n",
    "else:\n",
    "    pass"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Función para validar contenedores"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def validateID(containerID):\n",
    "    containeridre = re.search(r'\\w\\w\\w\\w\\d\\d\\d\\d\\d\\d\\d', str(containerID))\n",
    "    #nonvalid = []\n",
    "    if containeridre:\n",
    "        return(\"Container ID valid\")\n",
    "    else:\n",
    "        #nonvalid.append(str(containerID))\n",
    "        return(\"----- Container ID not valid: \" + str(containerID) + '-----')\n",
    "\n",
    "def propietario(containerID):\n",
    "    aux = list(str(containerID))\n",
    "    return str(aux[18]+aux[19]+aux[20])\n",
    "\n",
    "def categoria(containerID):\n",
    "    aux = list(str(containerID))\n",
    "    return str(aux[21])\n",
    "\n",
    "def numeroserie(containerID):\n",
    "    aux = list(str(containerID))\n",
    "    return str(aux[22]+aux[23]+aux[24]+aux[25]+aux[26]+aux[27])\n",
    "\n",
    "def digitocontrol(containerID):\n",
    "    aux = list(str(containerID))\n",
    "    return str(aux[28])\n",
    "\n",
    "propietarioudf = udf(propietario, StringType())\n",
    "categoriaudf = udf(categoria, StringType())\n",
    "numeroserieudf = udf(numeroserie, StringType())\n",
    "digitocontroludf = udf(digitocontrol, StringType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "data5 = sqlContext.sql(\"SELECT DISTINCT container_id FROM parquetFile \")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "data5col = data5.select(\"container_id\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(container_id='DXTQ1407119'), Row(container_id='VXNB1938296')]"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "data5col.rdd.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'Container ID valid'"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "validateID('DXTQ1407119')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " '----- Container ID not valid: Row(container_id=None)-----',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " \"----- Container ID not valid: Row(container_id='JMYG190Z978')-----\",\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " \"----- Container ID not valid: Row(container_id='GJFL14A2798')-----\",\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " \"----- Container ID not valid: Row(container_id='CTVU1506A832')-----\",\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " \"----- Container ID not valid: Row(container_id='DUKF166276')-----\",\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid',\n",
       " 'Container ID valid']"
      ]
     },
     "execution_count": 29,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "data5.rdd.map(lambda row: validateID(str(row))).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Contenedores que no tienen un código válido"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['----- Container ID not valid: Row(container_id=None)-----',\n",
       " \"----- Container ID not valid: Row(container_id='JMYG190Z978')-----\",\n",
       " \"----- Container ID not valid: Row(container_id='GJFL14A2798')-----\",\n",
       " \"----- Container ID not valid: Row(container_id='CTVU1506A832')-----\",\n",
       " \"----- Container ID not valid: Row(container_id='DUKF166276')-----\"]"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "data5.rdd.map(lambda row: validateID(str(row))).filter(lambda x: str(x) != 'Container ID valid').collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(container_id='DXTQ1407119'), Row(container_id='VXNB1938296')]"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "data5.take(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Creando un nuevo dataframe con las columnas deseadas y eliminando los contenedores no válidos"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------------+--------------------+---------------+\n",
      "|  ship_imo|container_id|               owner|container_group|\n",
      "+----------+------------+--------------------+---------------+\n",
      "|POG1615575| CDGP1673672|            Haag LLC|           22VH|\n",
      "|YZX1455509| RVNZ1574043|         Kuhic Group|           28VH|\n",
      "|NCZ1777367| MCLQ1045130|     Casper and Sons|           28VH|\n",
      "|FUS1202266| JNDM1523352|Reinger, VonRuede...|           28VH|\n",
      "|POG1615575| DLCR1100460|Hegmann, Hintz an...|           42T0|\n",
      "|NCZ1777367| HXNW1061510|       Sipes-Farrell|           42T0|\n",
      "|GEU1548633| CHTX1389862|      Homenick-Haley|           22VH|\n",
      "|MBV1836745| YBZA1370157|Weber, Senger and...|           42G0|\n",
      "|JCI1797526| FMHW1096477|        Grimes Group|           28VH|\n",
      "|GLV1922612| PJGZ1036450|Jerde, Collier an...|           28VH|\n",
      "|YZX1455509|IJWDR1216916|Mante, Cruickshan...|           28VH|\n",
      "|SQH1155999| DENF1086740|Kulas, Corkery an...|           42VH|\n",
      "|FUS1202266| NSIT1499568|     Dietrich-Legros|           28VH|\n",
      "|GEU1548633| LWJP1212809|    Keebler and Sons|           22T0|\n",
      "|MBV1836745| WZSO1549824|Becker, Cormier a...|           28VH|\n",
      "|GEU1548633| JMYG190Z978|           Huels LLC|           28VH|\n",
      "|JCI1797526| KITU1553521|     Schamberger LLC|           42T0|\n",
      "|NLH1771681| EQPB1510106|     Keebler-Smitham|           28VH|\n",
      "|POG1615575| QKOJ1756061|     Carter and Sons|           28VH|\n",
      "|NLH1771681| MTXR1152012| Runolfsson-MacGyver|           L5R1|\n",
      "+----------+------------+--------------------+---------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sqlContext.sql(\"SELECT DISTINCT ship_imo, container_id, owner, container_group FROM parquetFile \").show()\n",
    "data5 = sqlContext.sql(\"SELECT DISTINCT ship_imo, container_id, owner, container_group FROM parquetFile \")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------------+--------------------+---------------+---------+------------+-------------+-----------+\n",
      "|  ship_imo|container_id|               owner|container_group|categoria|numero_serie|digitocontrol|propietario|\n",
      "+----------+------------+--------------------+---------------+---------+------------+-------------+-----------+\n",
      "|POG1615575| CDGP1673672|            Haag LLC|           22VH|        0|           0|            0|          0|\n",
      "|YZX1455509| RVNZ1574043|         Kuhic Group|           28VH|        0|           0|            0|          0|\n",
      "|NCZ1777367| MCLQ1045130|     Casper and Sons|           28VH|        0|           0|            0|          0|\n",
      "|FUS1202266| JNDM1523352|Reinger, VonRuede...|           28VH|        0|           0|            0|          0|\n",
      "|POG1615575| DLCR1100460|Hegmann, Hintz an...|           42T0|        0|           0|            0|          0|\n",
      "|NCZ1777367| HXNW1061510|       Sipes-Farrell|           42T0|        0|           0|            0|          0|\n",
      "|GEU1548633| CHTX1389862|      Homenick-Haley|           22VH|        0|           0|            0|          0|\n",
      "|MBV1836745| YBZA1370157|Weber, Senger and...|           42G0|        0|           0|            0|          0|\n",
      "|JCI1797526| FMHW1096477|        Grimes Group|           28VH|        0|           0|            0|          0|\n",
      "|GLV1922612| PJGZ1036450|Jerde, Collier an...|           28VH|        0|           0|            0|          0|\n",
      "|YZX1455509|IJWDR1216916|Mante, Cruickshan...|           28VH|        0|           0|            0|          0|\n",
      "|SQH1155999| DENF1086740|Kulas, Corkery an...|           42VH|        0|           0|            0|          0|\n",
      "|FUS1202266| NSIT1499568|     Dietrich-Legros|           28VH|        0|           0|            0|          0|\n",
      "|GEU1548633| LWJP1212809|    Keebler and Sons|           22T0|        0|           0|            0|          0|\n",
      "|MBV1836745| WZSO1549824|Becker, Cormier a...|           28VH|        0|           0|            0|          0|\n",
      "|GEU1548633| JMYG190Z978|           Huels LLC|           28VH|        0|           0|            0|          0|\n",
      "|JCI1797526| KITU1553521|     Schamberger LLC|           42T0|        0|           0|            0|          0|\n",
      "|NLH1771681| EQPB1510106|     Keebler-Smitham|           28VH|        0|           0|            0|          0|\n",
      "|POG1615575| QKOJ1756061|     Carter and Sons|           28VH|        0|           0|            0|          0|\n",
      "|NLH1771681| MTXR1152012| Runolfsson-MacGyver|           L5R1|        0|           0|            0|          0|\n",
      "+----------+------------+--------------------+---------------+---------+------------+-------------+-----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "data5 = data5.withColumn('categoria', lit(0))\n",
    "data5 = data5.withColumn('numero_serie', lit(0))\n",
    "data5 = data5.withColumn('digitocontrol', lit(0))\n",
    "data5 = data5.withColumn('propietario', lit(0))\n",
    "data5.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Comprobamos los resultados de las funciones previamente creadas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['CDG']"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "data5.select('container_id').rdd.map(lambda row: propietario(row)).take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['167367']"
      ]
     },
     "execution_count": 35,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "data5.select('container_id').rdd.map(lambda row: numeroserie(row)).take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['P']"
      ]
     },
     "execution_count": 36,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "data5.select('container_id').rdd.map(lambda row: categoria(row)).take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['2']"
      ]
     },
     "execution_count": 37,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "data5.select('container_id').rdd.map(lambda row: digitocontrol(row)).take(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Aplicamos las udf creadas y formamos el dataframe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "data5 = data5.withColumn('propietarios', propietarioudf(data5.container_id))\n",
    "data5 = data5.withColumn('numero_serie', numeroserieudf(data5.container_id))\n",
    "data5 = data5.withColumn('categoria', categoriaudf(data5.container_id))\n",
    "data5 = data5.withColumn('digitocontrol', digitocontroludf(data5.container_id))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Ejercicio 6. \n",
    "#### Extrae una lista con peso total de cada barco, net_weight, sumando cada contenedor y agrupado por los campos ship_imo y container_group. Devuelve un DataFrame con la siguiente estructura: ship_imo, ship_name, container, total_net_weight.\n",
    "---"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "data6 = df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(ship_imo='AMC1861710', ship_name='Jayden', country='BD', departure='201602183', container_id='FCUK1755843', container_type='4960', container_group='28VH', net_weight='44804866.62', gross_weight='2240243.33', owner='Streich-Wilkinson', declared='Music, Tools, Automotive & Health', contact='octavia@stammbednar.name', customs_ok='true'),\n",
       " Row(ship_imo='POG1615575', ship_name='Lake Eribertoland', country='CR', departure='2016021611', container_id='PDXW1549639', container_type='28VH', container_group='8888', net_weight='16681047.32', gross_weight='500431.42', owner='Senger and Sons', declared='Movies & Jewelery', contact='cindy.dubuque@roberts.org', customs_ok='true')]"
      ]
     },
     "execution_count": 40,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "data6.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "data6new = data6.drop(\"country\", \"departure\", \"container_type\", \\\n",
    "                      \"container_group\", \"gross_weight\", \"owner\", \\\n",
    "                      \"declared\", \"customs_ok\", \"contact\", \"net_weight\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------------------+------------+\n",
      "|  ship_imo|         ship_name|container_id|\n",
      "+----------+------------------+------------+\n",
      "|AMC1861710|            Jayden| FCUK1755843|\n",
      "|POG1615575| Lake Eribertoland| PDXW1549639|\n",
      "|SQH1155999|            Aileen| PLKO1661930|\n",
      "|JCI1797526|          Herminio| BXMT1827488|\n",
      "|MBV1836745|Port Guiseppeburgh| JYIE1892741|\n",
      "|GYR1192020|         Emardland| LARQ1499256|\n",
      "|GLV1922612|           Eulalia| ARDX1463154|\n",
      "|NLH1771681|       Port Noemie| JFPX1246669|\n",
      "|FUS1202266|  East Mustafaland| ICAV1235470|\n",
      "|GLV1922612|           Eulalia| KEVU1145768|\n",
      "|IWE1254579|      North Creola| VDUQ1801278|\n",
      "|JET1053895|             Jamil| CXZN1286843|\n",
      "|KSP1096387|             Wiley| YAZN1142572|\n",
      "|GYR1192020|         Emardland| HVFU1799048|\n",
      "|GYR1192020|         Emardland| ROML1055099|\n",
      "|JMP1637582|East Zechariahland| LONM1299749|\n",
      "|TCU1641123|     New Margarete| XKAO1357085|\n",
      "|MBV1836745|Port Guiseppeburgh| JYPA1889923|\n",
      "|POG1615575| Lake Eribertoland| GKXC1181753|\n",
      "|AEY1108363|             Juana| GJFL14A2798|\n",
      "+----------+------------------+------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "data6new.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "data6new = data6new.withColumn(\"total_net_weight\", lit(0))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Test con la primera fila de datos"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "47045109.949999996"
      ]
     },
     "execution_count": 44,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "totalweight = sqlContext.sql(\"SELECT net_weight, gross_weight FROM parquetFile \")\n",
    "totalweight.rdd.take(1)\n",
    "float(totalweight.rdd.take(1)[0][0]) + float(totalweight.rdd.take(1)[0][1])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "PythonRDD[154] at RDD at PythonRDD.scala:48"
      ]
     },
     "execution_count": 45,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "totalweight.rdd.map(lambda x: float(x[0][0]) + float(x[0][1]))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Ejercicio 7.\n",
    "#### Guarda los resultados del ejercicio anterior en formato Parquet.\n",
    "---"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "df.write.parquet(\"ejercicios_6.parquet\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Ejercicio 8.\n",
    "#### ¿En qué casos crees que es más eficiente utilizar formatos como Parquet? ¿Existe alguna desventaja frente a formatos de texto como CSV?\n",
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Los formatos columnares facilitan el procesamiento, lectura y escritura de los motores como Spark, siendo hasta 10 veces más rápidos en leer los mismos datos almacenados en csv.\n",
    "\n",
    "Un ejemplo práctico en el que observar las ventajas del archivos como parquet es el caso de Amazon Athenea, el cual es un servicio para realizar consultas SQL. ['Ejemplo de precios'](https://aws.amazon.com/es/athena/pricing/)\n",
    "\n",
    "En el anterior enlace se explica como la misma consulta realizada sobre archivos parquet frente a csv permite reducir el coste de una consulta de 15€(sobre csv) a 1.67€(sobre parquet)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Ejercicio 9.\n",
    "#### ¿Es posible procesar XML mediante Spark? ¿Existe alguna restricción por la cual no sea eficiente procesar un único archivo en multiples nodos? ¿Se te ocurre alguna posible solución para trocear archivos suficientemente grandes? ¿Existe la misma problemática con otros formatos de texto como JSON?\n",
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Sería posible mediante un librería externa (spark.read.xml que viene de databricks)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Ejercicio 10.\n",
    "#### Spark SQL tiene una función denominada avg que se utiliza para calcular el promedio de un conjunto de valores ¿Por qué los autores han creado esta función en lugar de usar el API estándar de Python o Scala?\n",
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Para que pueda ser ejecutada de forma distribuida en un RDD."
   ]
  }
 ],
 "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
}