<jupyter_start><jupyter_text>SparkSQL y Spark DataFrames `SparkSQL` además de permitirnos interactuar usando `SQL` (en realidad `HQL` _Hive query language_, ver documentación [aquí](https://cwiki.apache.org/confluence/display/Hive/LanguageManual)), agrega una capa de abstracción al `RDD` y lo convierte en un `DataFrame` análogo al usado en `R` y `Python`.Se recomienda tomar como referencia el siguiente link : http://spark.apache.org/docs/2.4.3/api/python/pyspark.sql.html Como siempre, empezamos obtniendo el `SparkContext`<jupyter_code>import pyspark
sc = pyspark.SparkContext('local[*]')<jupyter_output><empty_output><jupyter_text>Pero además, obtendremos el `SQLContext` en esta ocasión<jupyter_code>sqlContext = pyspark.SQLContext(sc)<jupyter_output><empty_output><jupyter_text>Creando un DataFrame Ahora leeremos el archivo de transacciones creado en el laboratorio anterior<jupyter_code>! ls -lh output/raw/transacciones<jupyter_output>"ls" no se reconoce como un comando interno o externo,
programa o archivo por lotes ejecutable.<jupyter_text>**NOTA**: Si marca error la última instrucción, regresa al laboratorio anterior y crea de nuevo el archivo.<jupyter_code>txs_rdd = sc.textFile("output/raw/transacciones")
txs_rdd.first()
type(txs_rdd)<jupyter_output><empty_output><jupyter_text>Para poder obtener un `DataFrame` hay que extraer cada línea del `RDD` a un objeto `Row`<jupyter_code>from pyspark.sql import Row
Transaccion = Row('tdc', 'comercio', 'accion', 'monto')
def getTransaccion(linea):
    cells = linea.split('|')
    cells[3] = int(cells[3])
    return Transaccion(*cells)

txs = txs_rdd.map(getTransaccion)
txs.first()
txs_df = txs.toDF()
#para obtener un dataframe de spark
txs_df
txs
txs_df.printSchema()
txs_df.show()<jupyter_output>+--------------------+--------------------+------+-----+
|                 tdc|            comercio|accion|monto|
+--------------------+--------------------+------+-----+
|fc746fc2-3160-4f8...|RESTAURANTE EL TR...|COMPRA| 9274|
|e434d65e-cb57-43e...|RESTAURANTE EL TR...|COMPRA|  702|
|e434d65e-cb57-43e...|       ARENA COLISEO|COMPRA| 3070|
|8ef61de7-b609-4e6...|           SUPERCITO|COMPRA| 2128|
|fc746fc2-3160-4f8...|RESTAURANTE EL TR...|RETIRO| 4397|
|8ef61de7-b609-4e6...|RESTAURANTE EL TR...|RETIRO| 9042|
|cb679bdc-e5c4-476...|       ARENA COLISEO|RETIRO| 4549|
|8ef61de7-b609-4e6...|       ARENA COLISEO|RETIRO|  466|
|8ef61de7-b609-4e6...|           SUPERCITO|COMPRA| 8565|
|cb679bdc-e5c4-476...|       ARENA COLISEO|COMPRA| 1015|
|fc746fc2-3160-4f8...|RESTAURANTE EL TR...|COMPRA| 4119|
|e434d65e-cb57-43e...|       ARENA COLISEO|RETIRO| 8176|
|d03e74dd-0666-441...|RESTAURANTE EL TR...|RETIRO| 6224|
|8ef61de7-b609-4e6...|       ARENA COLISEO|COMPRA| 2085|
|8ef61de7-b609-4e6...|       AR[...]<jupyter_text>Registrando una tabla Al tener un `DataFrame` es posible usar la `API` o usar `SQL`. Para mostrarlo, primero registremos la tabla<jupyter_code>txs_df.registerTempTable('txs')
sqlContext.sql("show tables").show()
sqlContext.sql("select * from txs limit 5").show()<jupyter_output>+--------------------+--------------------+------+-----+
|                 tdc|            comercio|accion|monto|
+--------------------+--------------------+------+-----+
|fc746fc2-3160-4f8...|RESTAURANTE EL TR...|COMPRA| 9274|
|e434d65e-cb57-43e...|RESTAURANTE EL TR...|COMPRA|  702|
|e434d65e-cb57-43e...|       ARENA COLISEO|COMPRA| 3070|
|8ef61de7-b609-4e6...|           SUPERCITO|COMPRA| 2128|
|fc746fc2-3160-4f8...|RESTAURANTE EL TR...|RETIRO| 4397|
+--------------------+--------------------+------+-----+<jupyter_text>Ahora que `txs` ya tiene esquema, es buena idea guardarlo<jupyter_code>sqlContext.sql("select * from txs")\
          .coalesce(1)\
          .write.format("parquet")\
          .save("output/parquet/transacciones", mode="OVERWRITE")
sqlContext.sql("select * from txs")\
          .coalesce(1)\
          .write.format("json")\
          .save("output/json/transacciones", mode="OVERWRITE")<jupyter_output><empty_output><jupyter_text>Usando la API Al igual que los `RDD`s, los `DataFrames` pueden ser operados mediante _transformaciones_ y _acciones_. Las **Transformaciones** son _lazy_ pero contribuyen a la planeación de la ejecución del _query_, las **Acciones** provocan la ejecución del _query_  Transformaciones- `filter`- `select`- `drop`- `join` Acciones- `count`- `collect`- `show`- `head`- `take`<jupyter_code>txs_df.select(txs_df['tdc'], txs_df['accion'], txs_df['monto'], txs_df['monto'] >= 5000).show(5)
sqlContext.sql("select tdc, accion, monto, monto >= 5000 from txs").show(5)
txs_df.filter((txs_df["monto"] >= 5000) & (txs_df["accion"] == "RETIRO"))\
      .select(txs_df["tdc"], txs_df["comercio"], txs_df["monto"])\
      .orderBy(txs_df["monto"].desc())\
      .show(5)<jupyter_output>+--------------------+--------------------+-----+
|                 tdc|            comercio|monto|
+--------------------+--------------------+-----+
|cb679bdc-e5c4-476...|       ARENA COLISEO| 9996|
|fc746fc2-3160-4f8...|RESTAURANTE EL TR...| 9988|
|d03e74dd-0666-441...|RESTAURANTE EL TR...| 9986|
|d03e74dd-0666-441...|RESTAURANTE EL TR...| 9986|
|cb679bdc-e5c4-476...|RESTAURANTE EL TR...| 9984|
+--------------------+--------------------+-----+
only showing top 5 rows<jupyter_text>**NOTA**: `where()` es un _alias_ para `filter()`<jupyter_code>txs_df.where((txs_df["monto"] >= 5000) & (txs_df["accion"] == "RETIRO"))\
      .select(txs_df["tdc"], txs_df["comercio"], txs_df["monto"])\
      .show(5)<jupyter_output>+--------------------+--------------------+-----+
|                 tdc|            comercio|monto|
+--------------------+--------------------+-----+
|8ef61de7-b609-4e6...|RESTAURANTE EL TR...| 9042|
|e434d65e-cb57-43e...|       ARENA COLISEO| 8176|
|d03e74dd-0666-441...|RESTAURANTE EL TR...| 6224|
|e434d65e-cb57-43e...|           SUPERCITO| 9743|
|cb679bdc-e5c4-476...|       ARENA COLISEO| 7390|
+--------------------+--------------------+-----+
only showing top 5 rows<jupyter_text>También es posible usar `strings` para el condicional<jupyter_code>txs_df.filter("monto >= 5000 and accion = 'RETIRO'")\
      .select(txs_df["tdc"], txs_df["comercio"], txs_df["monto"])\
      .orderBy(txs_df["monto"].asc())\
      .show(5)
txs_df.groupBy("tdc").count().show(5)<jupyter_output>+--------------------+-----+
|                 tdc|count|
+--------------------+-----+
|cb679bdc-e5c4-476...| 1922|
|e434d65e-cb57-43e...| 2000|
|fc746fc2-3160-4f8...| 1987|
|8ef61de7-b609-4e6...| 2037|
|d03e74dd-0666-441...| 2054|
+--------------------+-----+<jupyter_text>Agregaciones<jupyter_code>from pyspark.sql.functions import sum, avg, max, min, round, count, col

# otra manera de escribir el count usando groupBy y agg
txs_df.groupBy("tdc").agg(count('tdc').alias('cant_tx')).show(5)
# Para cada comercio obtener el monto promedio de transacción (no importa si es RETIRO o COMPRA)
txs_df.groupBy('comercio').agg(avg('monto').alias('monto_promedio')).show()<jupyter_output>+--------------------+-----------------+
|            comercio|   monto_promedio|
+--------------------+-----------------+
|           SUPERCITO|4951.982918789332|
|RESTAURANTE EL TR...|4986.653122198984|
|       ARENA COLISEO|5074.468335343788|
+--------------------+-----------------+<jupyter_text>EjercicioPara cada comercio obtener y mostrar los máximos montos por acción.<jupyter_code>txs_df.groupby('comercio','accion').agg(max('monto')).show()<jupyter_output>+--------------------+------+----------+
|            comercio|accion|max(monto)|
+--------------------+------+----------+
|           SUPERCITO|COMPRA|      9999|
|           SUPERCITO|RETIRO|      9983|
|       ARENA COLISEO|RETIRO|      9996|
|RESTAURANTE EL TR...|COMPRA|      9986|
|RESTAURANTE EL TR...|RETIRO|      9988|
|       ARENA COLISEO|COMPRA|      9984|
+--------------------+------+----------+<jupyter_text>EjercicioObtener y mostrar los 10 clientes con mas transacciones por comercio ordenados de manera descendente<jupyter_code>df1 = txs_df.groupBy('tdc','comercio').agg(count('tdc').alias('cantidad_tx'))
df1.orderBy(df1['cantidad_tx'].desc()).show(10,truncate=False)<jupyter_output>+------------------------------------+----------------------+-----------+
|tdc                                 |comercio              |cantidad_tx|
+------------------------------------+----------------------+-----------+
|e434d65e-cb57-43ef-a16e-b3a6ee226144|RESTAURANTE EL TRABAJO|710        |
|d03e74dd-0666-4418-bec6-df04651842ba|ARENA COLISEO         |702        |
|d03e74dd-0666-4418-bec6-df04651842ba|SUPERCITO             |694        |
|8ef61de7-b609-4e65-9e0b-fd00eb8251a8|SUPERCITO             |688        |
|8ef61de7-b609-4e65-9e0b-fd00eb8251a8|ARENA COLISEO         |676        |
|8ef61de7-b609-4e65-9e0b-fd00eb8251a8|RESTAURANTE EL TRABAJO|673        |
|fc746fc2-3160-4f83-b53e-fe020a8c6f9a|SUPERCITO             |668        |
|fc746fc2-3160-4f83-b53e-fe020a8c6f9a|ARENA COLISEO         |661        |
|d03e74dd-0666-4418-bec6-df04651842ba|RESTAURANTE EL TRABAJO|658        |
|fc746fc2-3160-4f83-b53e-fe020a8c6f9a|RESTAURANTE EL TRABAJO|658        |
+------------------------------------+[...]<jupyter_text>User Defined Functions: UDF Supongamos que queremos mostrar el  `tdc` en mayúsculas, para tal menester es necesario definir una `UDF`<jupyter_code>from pyspark.sql.functions import udf
to_upper = udf(lambda s: s.upper())
txs_df.select(to_upper(txs_df["tdc"]).alias("Mayúsculas"), txs_df["tdc"])\
      .distinct()\
      .show(100)<jupyter_output>+--------------------+--------------------+
|          Mayúsculas|                 tdc|
+--------------------+--------------------+
|8EF61DE7-B609-4E6...|8ef61de7-b609-4e6...|
|CB679BDC-E5C4-476...|cb679bdc-e5c4-476...|
|D03E74DD-0666-441...|d03e74dd-0666-441...|
|FC746FC2-3160-4F8...|fc746fc2-3160-4f8...|
|E434D65E-CB57-43E...|e434d65e-cb57-43e...|
+--------------------+--------------------+<jupyter_text>EjercicioUsando una udf, generar un nuevo dataframe (txs_df2) con una nueva columna (prefijo_tdc) que tenga el primer segmento de tdc y mostrarlo<jupyter_code>cortar = lambda s: s.split('-')[0] #nos quedamos con el primer elemento
pref_tdc = udf(lambda s: s.split('-')[0])
txs_df2 = txs_df.select ('*',pref_tdc(txs_df['tdc']).alias('prefijo_tdc'))
txs_df2.show(10)<jupyter_output>+--------------------+--------------------+------+-----+-----------+
|                 tdc|            comercio|accion|monto|prefijo_tdc|
+--------------------+--------------------+------+-----+-----------+
|fc746fc2-3160-4f8...|RESTAURANTE EL TR...|COMPRA| 9274|   fc746fc2|
|e434d65e-cb57-43e...|RESTAURANTE EL TR...|COMPRA|  702|   e434d65e|
|e434d65e-cb57-43e...|       ARENA COLISEO|COMPRA| 3070|   e434d65e|
|8ef61de7-b609-4e6...|           SUPERCITO|COMPRA| 2128|   8ef61de7|
|fc746fc2-3160-4f8...|RESTAURANTE EL TR...|RETIRO| 4397|   fc746fc2|
|8ef61de7-b609-4e6...|RESTAURANTE EL TR...|RETIRO| 9042|   8ef61de7|
|cb679bdc-e5c4-476...|       ARENA COLISEO|RETIRO| 4549|   cb679bdc|
|8ef61de7-b609-4e6...|       ARENA COLISEO|RETIRO|  466|   8ef61de7|
|8ef61de7-b609-4e6...|           SUPERCITO|COMPRA| 8565|   8ef61de7|
|cb679bdc-e5c4-476...|       ARENA COLISEO|COMPRA| 1015|   cb679bdc|
+--------------------+--------------------+------+-----+-----------+
only showing top 10 rows<jupyter_text>Imaginemos que sólo estamos interesados en aquellas transacciones que fueron `RETIRO`s en el `SUPERCITO` por montos mayores a `9000` ya que resultan sospechosos.<jupyter_code>txs_sospechosas = txs_df.filter("monto >= 9000 and accion = 'RETIRO' and comercio = 'SUPERCITO'")\
      .select(txs_df["tdc"],  txs_df["monto"])\
      .orderBy(txs_df["monto"].desc())
txs_sospechosas.count()<jupyter_output><empty_output><jupyter_text>Guardemos la información para un posterior análisis<jupyter_code>txs_sospechosas.coalesce(1)\
               .write.format("json")\
               .save("output/json/transacciones_sospechosas", mode="OVERWRITE")
txs_sospechosas.coalesce(1)\
               .write.format("parquet")\
               .save("output/parquet/transacciones_sospechosas", mode="OVERWRITE")
! ls -lh output/json/transacciones_sospechosas<jupyter_output><empty_output><jupyter_text>DataFrame desde JSON Para esta sección usaremos los datos de proyectos soportados por el **Banco Mundial**<jupyter_code>! rm -R data/world_bank*
! wget https://github.com/bradenrc/sparksql_pot/raw/master/world_bank.zip -P data/
! unzip data/world_bank.zip -d data/world_bank
! rm data/world_bank.zip<jupyter_output><empty_output><jupyter_text>Creamos el `DataFrame`<jupyter_code>world_bank = sqlContext.read.json("world_bank.json")<jupyter_output><empty_output><jupyter_text>Automáticamente detecta el _esquema_ de la fuente de datos<jupyter_code>world_bank.printSchema()<jupyter_output>root
 |-- _id: struct (nullable = true)
 |    |-- $oid: string (nullable = true)
 |-- approvalfy: string (nullable = true)
 |-- board_approval_month: string (nullable = true)
 |-- boardapprovaldate: timestamp (nullable = true)
 |-- borrower: string (nullable = true)
 |-- closingdate: timestamp (nullable = true)
 |-- country_namecode: string (nullable = true)
 |-- countrycode: string (nullable = true)
 |-- countryname: string (nullable = true)
 |-- countryshortname: string (nullable = true)
 |-- docty: string (nullable = true)
 |-- envassesmentcategorycode: string (nullable = true)
 |-- grantamt: long (nullable = true)
 |-- ibrdcommamt: long (nullable = true)
 |-- id: string (nullable = true)
 |-- idacommamt: long (nullable = true)
 |-- impagency: string (nullable = true)
 |-- lendinginstr: string (nullable = true)
 |-- lendinginstrtype: string (nullable = true)
 |-- lendprojectcost: long (nullable = true)
 |-- majorsector_percent: array (nullable = true)
 |    |-- element: struct (cont[...]<jupyter_text>La estructura es anidada (en lugar de bidimensional), pero observa que `Spark` no tuvo ningún problema.<jupyter_code>world_bank.registerTempTable("world_bank_projects")
sqlContext.sql('show tables').show()
sqlContext.sql('select countryshortname, project_name, totalamt, totalcommamt from world_bank_projects order by countryshortname').show()
projects_by_country = sqlContext.sql('select countryshortname as country, count(project_name) as num_projects, sum(totalamt) as total_amount from world_bank_projects group by countryshortname order by total_amount desc')
projects_by_country.show()
projects_by_country.printSchema()<jupyter_output>root
 |-- country: string (nullable = true)
 |-- num_projects: long (nullable = false)
 |-- total_amount: long (nullable = true)<jupyter_text>EjercicioA partir del dataframe projects_by_country encontrar el monto promedio por proyecto para Uruguay y Bolivia<jupyter_code>projects_by_country.filter((projects_by_country['country'] == 'Uruguay') | (projects_by_country['country'] == 'Bolivia')).select('country', (col('total_amount') / col('num_projects')).alias('promedio_por_pais')).show()<jupyter_output>+-------+-----------------+
|country|promedio_por_pais|
+-------+-----------------+
|Uruguay|           1.22E8|
|Bolivia|            3.7E7|
+-------+-----------------+<jupyter_text>EjercicioA partir del dataframe projects_by_country encontrar los 5 paises con mayores montos promedio<jupyter_code>df = projects_by_country
df1 = df.select('country', (col('total_amount') / col('num_projects')).alias('promedio_por_pais'))
df1.orderBy(df1['promedio_por_pais'].desc()).show(5,truncate=False)<jupyter_output>+-----------------------+-------------------+
|country                |promedio_por_pais  |
+-----------------------+-------------------+
|Poland                 |1.3078E9           |
|Turkey                 |3.2525E8           |
|Colombia               |3.166666666666667E8|
|Ethiopia               |3.1125E8           |
|Egypt, Arab Republic of|2.927E8            |
+-----------------------+-------------------+
only showing top 5 rows<jupyter_text>SparkSQL y Pandas Es posible usar `Pandas` para hacer análisis, pero hay que tomar en cuenta que esto manda _todo_ el `dataset` a un sólo nodo.<jupyter_code>import pandas as pd
projects_by_country_pd = projects_by_country.toPandas()
projects_by_country_pd.columns
projects_by_country_pd=projects_by_country_pd.set_index(['country'])
projects_by_country_pd.num_projects
projects_by_country_pd.head()
projects_by_country_pd.loc['Peru']
projects_by_country_pd['num_projects'][:10].plot(kind='barh', rot=0, )<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.22921485088253196, 'fraction_numerical': 0.06944613511868533, 'mean_word_length': 2.7608148317692836, 'pattern_counts': {'":': 0, '<': 82, '<?xml version=': 0, '>': 88, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 1, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '24', 'dir': 'jupyter-structured-clean-dedup', 'id': '405699', 'license': 'MIT', 'n_tokens_mistral': 6801, 'n_tokens_neox': 6166, 'n_words': 1049, 'path': 'dataframes_II (1).ipynb', 'repo_name': 'road2datascience/road2datascience.github.io', 'size': '16430'}