<jupyter_start><jupyter_text>SparkSession & SparkContext En versiones de Spark anteriores a 2.0.0, __sparkContext__ era usado como el canal para acceder a todas las funcionalidades de spark. Se encarga de gestionar la conexión con el resource manager y los ejecutores.A partir de la versión 2.0.0, __sparkSession__ funciona como punto de entrada único para interactuar con spark, contiene toda la funcionalidad básica del _sparkContext_ y añade la API para interacturar con Dataframes. En el caso de Intelligence ambos objetos son creados automáticamente al iniciar un _kernel_ de _pyspark_.<jupyter_code>spark
spark.version
sc<jupyter_output><empty_output><jupyter_text>_sparkSession_ contiene al _sparkContext_<jupyter_code>spark.sparkContext<jupyter_output><empty_output><jupyter_text>_sparkContext_ contiene la configuración de spark<jupyter_code>sc.getConf().getAll()<jupyter_output><empty_output><jupyter_text>DataFramesEs una colección inmutable y distribuida de datos organizados en columnas de forma tabular que permite una abstracción de los datos a alto nivel.__Características__- Conjunto de filas (Rows) con un esquema de datos (Schema)- Colección distribuida de datos organizados en filas y columnas nombradas.- Conceptualmente equivalente a una tabla en una BD relacional, o a un data frame de R o Python (pandas), pero con optimizaciones avanzadas para soportar aplicaciones Big Data y Data Science.- Pueden construirse a partir de distintas fuentes: ficheros de datos estructurados, tablas de Hive, BD externas, o RDDs- DataFrame API disponible en Scala, Java, Python, y R En SparkSQL cada __transformación__ sobre un DataFrame se añade a lo que se conoce como *query plan*. Cuando se aplica una __acción__ sobre el DataFrame el *Catalyst Optimizer* analiza el *query plan* e intenta optimizarlo, luego selecciona el plan físico (transformación a operaciones RDD de bajo nivel) más eficiente para la ejecución del plan, y lo ejecuta. Se puede consultar el *query plan* usando la función `explain()` sobre el DataFrame.  Creación de DataFramesNormalmente un DataFrame se crea leyendo datos desde una fuente externa (S3, HDFS, etc). Por ejemplo para leer el fichero 'Building_Permits.csv' almacenado en el Sandbox Data:<jupyter_code>file_name = "Building_Permits.csv"

buildings_df = spark.read.csv(DATA_PATH + file_name, sep=',', header=True, inferSchema=True)
buildings_df.show(3)<jupyter_output>+-------------+-----------+----------------------+--------------------+-----+---+-------------+--------------------+-----------+-------------+----+-----------+--------------------+--------------+-------------------+----------+-----------+--------------+--------------------------------+-----------------------+--------------------------+--------------------------+-----------------------------+----------------+----------------------+--------------+------------+-------------------+--------------+------------+--------------+--------+---------------+--------------------------+--------------------------------------+--------------------------+--------------------------------------+-----------+-------------------+-----------------------------------+-------+--------------------+-------------+
|Permit Number|Permit Type|Permit Type Definition|Permit Creation Date|Block|Lot|Street Number|Street Number Suffix|Street Name|Street Suffix|Unit|Unit Suffix|         Description|Current Status|Current Sta[...]<jupyter_text>En el caso de leer ficheros parquet no es necesario indicar una cabecera ni separador. Esta iformación ya está contenida en los propios ficheros parquet. Por ejemplo:<jupyter_code>file_path = "parquet/online_retail"

online_df = spark.read.parquet(DATA_PATH + file_path)
online_df.show(5)<jupyter_output>+---------+---------+--------------------+--------+---------------+---------+----------+--------------+
|InvoiceNo|StockCode|         Description|Quantity|    InvoiceDate|UnitPrice|CustomerID|       Country|
+---------+---------+--------------------+--------+---------------+---------+----------+--------------+
|   536365|   85123A|WHITE HANGING HEA...|       6|01/12/2010 8:26|     2,55|     17850|United Kingdom|
|   536365|    71053| WHITE METAL LANTERN|       6|01/12/2010 8:26|     3,39|     17850|United Kingdom|
|   536365|   84406B|CREAM CUPID HEART...|       8|01/12/2010 8:26|     2,75|     17850|United Kingdom|
|   536365|   84029G|KNITTED UNION FLA...|       6|01/12/2010 8:26|     3,39|     17850|United Kingdom|
|   536365|   84029E|RED WOOLLY HOTTIE...|       6|01/12/2010 8:26|     3,39|     17850|United Kingdom|
+---------+---------+--------------------+--------+---------------+---------+----------+--------------+
only showing top 5 rows<jupyter_text>También se puede crear un DataFrame de Spark a partir de un DataFrame de Pandas.<jupyter_code>import pandas as pd

df = pd.DataFrame({'city': ['Madrid', 'Birmingham', 'Barcelona', 'Mexico City'], 
                   'population': [3.2, 0.2, 1.6, 8.8]})
df
df_s = spark.createDataFrame(df)
df_s
df_s.show()<jupyter_output>+-----------+----------+
|       city|population|
+-----------+----------+
|     Madrid|       3.2|
| Birmingham|       0.2|
|  Barcelona|       1.6|
|Mexico City|       8.8|
+-----------+----------+<jupyter_text>Información básica de DataFrames  Previsualización`show` es un método que muestra por pantalla _n_ filas del DataFrame.<jupyter_code>online_df.show(6)<jupyter_output>+---------+---------+--------------------+--------+---------------+---------+----------+--------------+
|InvoiceNo|StockCode|         Description|Quantity|    InvoiceDate|UnitPrice|CustomerID|       Country|
+---------+---------+--------------------+--------+---------------+---------+----------+--------------+
|   536365|   85123A|WHITE HANGING HEA...|       6|01/12/2010 8:26|     2,55|     17850|United Kingdom|
|   536365|    71053| WHITE METAL LANTERN|       6|01/12/2010 8:26|     3,39|     17850|United Kingdom|
|   536365|   84406B|CREAM CUPID HEART...|       8|01/12/2010 8:26|     2,75|     17850|United Kingdom|
|   536365|   84029G|KNITTED UNION FLA...|       6|01/12/2010 8:26|     3,39|     17850|United Kingdom|
|   536365|   84029E|RED WOOLLY HOTTIE...|       6|01/12/2010 8:26|     3,39|     17850|United Kingdom|
|   536365|    22752|SET 7 BABUSHKA NE...|       2|01/12/2010 8:26|     7,65|     17850|United Kingdom|
+---------+---------+--------------------+--------+-------------[...]<jupyter_text>DimensionsEn Spark, no existe un método *shape*, por lo que hay que contar por separados las filas y las columnas.<jupyter_code>buildings_df.count()<jupyter_output><empty_output><jupyter_text>`columns` es un atributo que contiene los nombres de las columnas del DataFrame.<jupyter_code>buildings_df.columns
len(buildings_df.columns)<jupyter_output><empty_output><jupyter_text>SchemaEl schema de un dataframe nos muestra como se interpretaran los datos. Esto no significa que los datos estén así. _schema_ es un atributo del objeto, no un método. _printSchema()_ es un método que muestra una versión más entendidible del _schema_.<jupyter_code>buildings_df.schema
buildings_df.printSchema()<jupyter_output><empty_output><jupyter_text>dtypesEl atributo `dtypes` contiene los nombres de las columnas del dataframe junto con su tipo. Esto permite seleccionar nombres de columnas basados en el tipo, normalmente las variables categóricas (string y boolean) tienen tratamientos distintos a las numéricas (enteras y decimales).<jupyter_code>online_df.dtypes
categorial_columns = [c for c,t in online_df.dtypes if t in ['string', 'bool']]
categorial_columns
numerical_columns = [c for c,t in online_df.dtypes if t in ['int', 'double']]
numerical_columns<jupyter_output><empty_output><jupyter_text>AccionesLas acciones principales de Spark que se van a tratar son las siguientes:        - show()    - count()    - first()    - take()    - collect()    - toPandas()    - write()  showMuestra por pantalla _n_ filas del DataFrame (20 por defecto). Es una llamada a un `print`, no permite almacenar el resultado en una variable. El parámetro `truncate` limita el número de caracteres de cada campo.<jupyter_code>online_df.show(5)
online_df.show(5, truncate=False)<jupyter_output>+---------+---------+-----------------------------------+--------+---------------+---------+----------+--------------+
|InvoiceNo|StockCode|Description                        |Quantity|InvoiceDate    |UnitPrice|CustomerID|Country       |
+---------+---------+-----------------------------------+--------+---------------+---------+----------+--------------+
|536365   |85123A   |WHITE HANGING HEART T-LIGHT HOLDER |6       |01/12/2010 8:26|2,55     |17850     |United Kingdom|
|536365   |71053    |WHITE METAL LANTERN                |6       |01/12/2010 8:26|3,39     |17850     |United Kingdom|
|536365   |84406B   |CREAM CUPID HEARTS COAT HANGER     |8       |01/12/2010 8:26|2,75     |17850     |United Kingdom|
|536365   |84029G   |KNITTED UNION FLAG HOT WATER BOTTLE|6       |01/12/2010 8:26|3,39     |17850     |United Kingdom|
|536365   |84029E   |RED WOOLLY HOTTIE WHITE HEART.     |6       |01/12/2010 8:26|3,39     |17850     |United Kingdom|
+---------+---------+---------------------------[...]<jupyter_text>countCuenta el número de filas del DataFrame<jupyter_code>online_df.count()
buildings_df.count()<jupyter_output><empty_output><jupyter_text>firstDevuelve una única fila del DataFrame como un objeto de tipo `Row`. A los elementos del objeto `Row` se puede acceder tanto por nombre como por posición.<jupyter_code>single_row = online_df.first()
single_row
type(single_row)
single_row[1]
single_row['StockCode']<jupyter_output><empty_output><jupyter_text>takeDevuelve _n_ filas del DataFrame como una lista de objetos `Row`.<jupyter_code>row_list = online_df.take(4)
row_list
row_list[1]
row_list[1][2]
row_list[1]['Description']<jupyter_output><empty_output><jupyter_text>collectVuelca en un solo nodo todos los datos y los almacena en formato lista de Rows. Funciona como un `take` sin límite. Es una función a evitar salvo en ocasiones muy específicas donde sea necesario tener todos los datos en local y no haya otra forma de gestionarlos. Utilizar únicamente con datos filtrados y/o agregados.<jupyter_code>all_rows = online_df.collect()
len(all_rows)
all_rows[:4]<jupyter_output><empty_output><jupyter_text>toPandasVuelca en un nodo todos los datos como un DataFrame de Pandas. Se deben seguir las mismas restricciones que con `collect`.<jupyter_code>online_pandas = online_df.toPandas()
type(online_pandas)
online_pandas.head(3)<jupyter_output><empty_output><jupyter_text>writeGuarda la información de la tabla en fichero. Se puede escribir tanto en texto plano (CSV) como en formato parquet.<jupyter_code>online_df.write.csv(DATA_PATH + 'online_retail_csv')
online_df.write.parquet(DATA_PATH + 'online_retail_parquet')<jupyter_output><empty_output><jupyter_text>Podemos comprobar que usando la misma ruta que hemos utilizado para escribir, podemos leer el DataFrame tanto de un CSV como de un Parquet<jupyter_code>spark.read.csv(DATA_PATH + "online_retail_csv").show(5)
spark.read.parquet(DATA_PATH + "online_retail_parquet").show(5)<jupyter_output>+---------+---------+--------------------+--------+---------------+---------+----------+--------------+
|InvoiceNo|StockCode|         Description|Quantity|    InvoiceDate|UnitPrice|CustomerID|       Country|
+---------+---------+--------------------+--------+---------------+---------+----------+--------------+
|   536365|   85123A|WHITE HANGING HEA...|       6|01/12/2010 8:26|     2,55|     17850|United Kingdom|
|   536365|    71053| WHITE METAL LANTERN|       6|01/12/2010 8:26|     3,39|     17850|United Kingdom|
|   536365|   84406B|CREAM CUPID HEART...|       8|01/12/2010 8:26|     2,75|     17850|United Kingdom|
|   536365|   84029G|KNITTED UNION FLA...|       6|01/12/2010 8:26|     3,39|     17850|United Kingdom|
|   536365|   84029E|RED WOOLLY HOTTIE...|       6|01/12/2010 8:26|     3,39|     17850|United Kingdom|
+---------+---------+--------------------+--------+---------------+---------+----------+--------------+
only showing top 5 rows<jupyter_text>La partición de tablas es un forma habitual de optimización. En una tabla particionada, los datos se almacenan en diferentes carpetas, donde los valores de la columna utilizada para particionar forman parte de la ruta de cada carpeta. Al leer un fichero particionado, Spark es capaz de descubrir e inferir la información de partición automáticamente para recuperar los datos correctamente.<jupyter_code>online_df.write.partitionBy("Country").parquet(DATA_PATH + "online_retail_parquet_partition")<jupyter_output><empty_output><jupyter_text>Podemos leer los datos del parquet que acabamos de escribir:Leyendo todo el parquet completo (sin especificar la columna con la que hemos particionado):<jupyter_code>spark.read.parquet(DATA_PATH + "online_retail_parquet_partition").show(5)<jupyter_output>+---------+---------+--------------------+--------+---------------+---------+----------+--------------+
|InvoiceNo|StockCode|         Description|Quantity|    InvoiceDate|UnitPrice|CustomerID|       Country|
+---------+---------+--------------------+--------+---------------+---------+----------+--------------+
|   536365|   85123A|WHITE HANGING HEA...|       6|01/12/2010 8:26|     2,55|     17850|United Kingdom|
|   536365|    71053| WHITE METAL LANTERN|       6|01/12/2010 8:26|     3,39|     17850|United Kingdom|
|   536365|   84406B|CREAM CUPID HEART...|       8|01/12/2010 8:26|     2,75|     17850|United Kingdom|
|   536365|   84029G|KNITTED UNION FLA...|       6|01/12/2010 8:26|     3,39|     17850|United Kingdom|
|   536365|   84029E|RED WOOLLY HOTTIE...|       6|01/12/2010 8:26|     3,39|     17850|United Kingdom|
+---------+---------+--------------------+--------+---------------+---------+----------+--------------+
only showing top 5 rows<jupyter_text>Leyendo solo la partición correspondiente a un valor de la columna utilizada para particionar:<jupyter_code>spark.read.parquet(DATA_PATH + "online_retail_parquet_partition/Country=United Kingdom").show(5)<jupyter_output>+---------+---------+--------------------+--------+---------------+---------+----------+
|InvoiceNo|StockCode|         Description|Quantity|    InvoiceDate|UnitPrice|CustomerID|
+---------+---------+--------------------+--------+---------------+---------+----------+
|   536365|   85123A|WHITE HANGING HEA...|       6|01/12/2010 8:26|     2,55|     17850|
|   536365|    71053| WHITE METAL LANTERN|       6|01/12/2010 8:26|     3,39|     17850|
|   536365|   84406B|CREAM CUPID HEART...|       8|01/12/2010 8:26|     2,75|     17850|
|   536365|   84029G|KNITTED UNION FLA...|       6|01/12/2010 8:26|     3,39|     17850|
|   536365|   84029E|RED WOOLLY HOTTIE...|       6|01/12/2010 8:26|     3,39|     17850|
+---------+---------+--------------------+--------+---------------+---------+----------+
only showing top 5 rows
{'fraction_non_alphanumeric': 0.2508526717046746, 'fraction_numerical': 0.06741122182839564, 'mean_word_length': 2.149536647009267, 'pattern_counts': {'":': 0, '<': 81, '<?xml version=': 0, '>': 81, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '22', 'dir': 'jupyter-structured-clean-dedup', 'id': '491649', 'license': 'MIT', 'n_tokens_mistral': 5334, 'n_tokens_neox': 4616, 'n_words': 1221, 'path': '2021Q1_DSF/5.- Spark/notebooks/spark_sql/01_spark_sql_basics.ipynb', 'repo_name': 'serch86/binder-pyspark-DSF_2021Q1', 'size': '14953'}