<jupyter_start><jupyter_text>Punto de entrada Vamos a crear un punto de entrada al API de dataframes y dataset.<jupyter_code>from pyspark.sql import SparkSession

spark= SparkSession.builder.appName("Trabajando con Spark SQL").getOrCreate()<jupyter_output><empty_output><jupyter_text>Lo primero que vamos a leer va a ser un fichero json que representa la tabla periodica y lo vamos a almacenar en un dataframe sobre el que vamos a ir realizando diferentes acciones como si se tratara de un RDD.Nota: el formato json en spark SQL es un formato por línea , como si fuera un CSV, por lo tanto, hay que transformar el listado de objetos en un una fila por cada objeto.<jupyter_code>import json
with open('sql/PeriodicTableJSON.json') as data_file:    
    data = json.load(data_file)
    

    with open('sql/PeriodicTableJSON.jsonl', 'w') as outfile:
        for entry in data:
            json.dump(entry, outfile)
            outfile.write('\n')
df = spark.read.json("sql/PeriodicTableJSON.jsonl")
df.show()
df.printSchema()
df.select("name").show()<jupyter_output>+----------+
|      name|
+----------+
|  Hydrogen|
|    Helium|
|   Lithium|
| Beryllium|
|     Boron|
|    Carbon|
|  Nitrogen|
|    Oxygen|
|  Fluorine|
|      Neon|
|    Sodium|
| Magnesium|
| Aluminium|
|   Silicon|
|Phosphorus|
|    Sulfur|
|  Chlorine|
|     Argon|
| Potassium|
|   Calcium|
+----------+
only showing top 20 rows<jupyter_text>Seleccionamos los elementos químicos que tengan la masa atómica menor que 200 y mostramos los 10 primeros.<jupyter_code>df.select(df['name'],df['atomic_mass']).filter(df['atomic_mass']<200).show(10)
df.groupBy('phase').count().show()<jupyter_output>+------+-----+
| phase|count|
+------+-----+
| Solid|  104|
|Liquid|    2|
|   Gas|   12|
+------+-----+<jupyter_text>Ahora vamos a ver como a partir de un dataframe podemos generar una tabla temporal sobre la que ejecutaremos sentencias en SQL.<jupyter_code>df.createGlobalTempView("chemistryTable")
spark.sql("select name from global_temp.chemistryTable").show(5)<jupyter_output>+---------+
|     name|
+---------+
| Hydrogen|
|   Helium|
|  Lithium|
|Beryllium|
|    Boron|
+---------+
only showing top 5 rows<jupyter_text>Como ya hemos comentado en el post, python no permite construir estructuras de dataset. Para que te hagas una idea si vienes del mundo Java o Scala. La creación de dataset se basa en la definición de una clase y permite añadir objetos de esa clase. El resultado es una estructura en formato de tabla como el dataframe mostrado en nuestro caso.  Infiriendo el esquema En Spark SQL, existen dos formas de inferir el esquema un dataframe. Una es mediante reflexión y la otra es explicitamente con programación. A continuación vamos a ver ambos casos sobre un documento txt que contiene el elemento químico y su masa atómica.<jupyter_code>from pyspark.sql import Row
sc = spark.sparkContext
lines=sc.textFile("sql/Periodictable.txt")
parts= lines.map(lambda p: p.split(","))

elements= parts.map(lambda e: Row(name=e[0],atomic_mass=float(e[1])))
schemeElements=spark.createDataFrame(elements)
schemeElements.createOrReplaceTempView("elements")
lightElements=spark.sql("select name from elements where atomic_mass>0 and atomic_mass<21")
lightElemName=lightElements.rdd.map(lambda elem: "Name: "+elem.name).collect()
for name in lightElemName:
    print(name)<jupyter_output>Name: Hydrogen
Name: Helium
Name: Lithium
Name: Beryllium
Name: Boron
Name: Carbon
Name: Nitrogen
Name: Oxygen
Name: Fluorine
Name: Neon<jupyter_text>Ahora vamos a ver como se haría programáticamente.<jupyter_code>from pyspark.sql.types import *

sc=spark.sparkContext

lines=sc.textFile("sql/Periodictable.txt")
parts=lines.map(lambda line: line.split(","))
elements= parts.map(lambda p: (p[0],p[1]))

schemeString="name atomicMass"

fields= [StructField(field_name,StringType(),True) for field_name in schemeString.split()]
scheme =StructType(fields)

schemeElements= spark.createDataFrame(elements,scheme)

schemeElements.createOrReplaceTempView("elements")

spark.sql("select name,atomicMass from elements").show()

schemeElements.printSchema()<jupyter_output>+---------+-------------+
|     name|   atomicMass|
+---------+-------------+
| Hydrogen|        1.008|
|   Helium|    4.0026022|
|  Lithium|         6.94|
|Beryllium|   9.01218315|
|    Boron|        10.81|
|   Carbon|       12.011|
| Nitrogen|       14.007|
|   Oxygen|       15.999|
| Fluorine|18.9984031636|
|     Neon|     20.17976|
+---------+-------------+

root
 |-- name: string (nullable = true)
 |-- atomicMass: string (nullable = true)<jupyter_text>Data Source Existen multitud de formatos disponible en Spark SQL (json,parquet,jdbc,orc,libsvm,csv,text,...) aunque el formato por defecto es parquet.En este apartado vamos a ver el manejo de diferentes formatos de datos y la comunicación con Hive, Parquet y JDBC para guardar/recuperar información. Lo primero que vamos hacer es escribir la query anterior a un fichero parquet para despues hacer queries directamente sobre el fichero.<jupyter_code>schemeElements.select("name","atomicMass").write.save("sql/namesAndAtomicMass.parquet")<jupyter_output><empty_output><jupyter_text>Y ahra podemos hacer una query directamente desde el fichero.<jupyter_code>newDf= spark.sql("select atomicMass from parquet.`sql/namesAndAtomicMass.parquet`")
newDf.show()<jupyter_output>+-------------+
|   atomicMass|
+-------------+
|        1.008|
|    4.0026022|
|         6.94|
|   9.01218315|
|        10.81|
|       12.011|
|       14.007|
|       15.999|
|18.9984031636|
|     20.17976|
+-------------+<jupyter_text>Cuando has arrancado el cluster de docker, en la base de datos hemos añadido una setencia books.sql alojada en notebook/initdb que carga dentro del esquema postgres la tabla libros con title author y año<jupyter_code>jdbcDF = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:postgresql://db:5432/postgres") \
    .option("dbtable", "books") \
    .option("user", "postgres") \
    .option("driver", "org.postgresql.Driver") \
    .option("password", "<PASSWORD>") \
    .load()
jdbcDF.show();
jdbcDF.printSchema();<jupyter_output>root
 |-- title: string (nullable = true)
 |-- author: string (nullable = true)
 |-- year: integer (nullable = true)
{'fraction_non_alphanumeric': 0.14157591287636132, 'fraction_numerical': 0.024823830877642537, 'mean_word_length': 3.87890625, 'pattern_counts': {'":': 0, '<': 33, '<?xml version=': 0, '>': 32, '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': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '652482', 'license': 'Apache-2.0', 'n_tokens_mistral': 2249, 'n_tokens_neox': 2137, 'n_words': 581, 'path': 'notebook/Trabajando con Spark SQL y dataframes.ipynb', 'repo_name': 'grecomariano/transporte', 'size': '6238'}