Vamos a hacer un ejemplo con Apache Spark SQL. Algo simple, vamos a tomar algunas tablas de una base de datos relacional y vamos hacer unas consultas y luego vamos a importar datos de un csv para tambien hacer consultas:
//Primero importamos la clase de SQLContext y creamos el contexto.
import org.apache.spark.sql.SQLContext
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
//Vamos a importar los datos de la base de datos, en este caso la tabla "actor"
val url = "jdbc:mysql://0.0.0.0:3306/sakila"
val dfActor = spark.read.format("jdbc").option("url", url)
.option("dbtable", "actor")
.option("user", "sparkDb")
.option("password","sparkDb").load()
//Ya tenemos la tabla en el dataframe ahora vamos a imprimir el esquema y ver los datos.
dfActor.printSchema()
dfActor.show()
dfActor.count()
//Consultamos el campo nombre
dfActor.select("first_name").show()
//Registramos la tabla para poder consultarla con SQL
dfActor.registerTempTable("actors")
// Y consultamos
spark.sql("select first_name, last_name from actors").show
//Vamos a traernos una tabla relación entre el actor y el film
val dfActorFilms = spark.read.format("jdbc").option("url", url).option("dbtable", "film_actor").option("user", "sparkDb").option("password","sparkDb").load()
//Registro la tabla temporal
dfActorFilms.registerTempTable("actors_films")
//Selecciono los datos del actor y en cuantas peliculas actuo
spark.sql("select a.first_name, a.last_name, count(af.film_id) from actors a join actors_films af on (a.actor_id=af.actor_id) group by a.first_name, a.last_name").show
// Ahora vamos a traernos los datos del film pero desde un archivo csv
// Creamos una clase que me va hacer util leer los datos de forma estructurada.
case class Film(film_id: Int, title: String, description: String, release_year: Int)
// Leemos el archivo CSV :
val dfFilms = sc.textFile("/home/spark/films.csv").map(_.split(",")).map(p => Film(p(0).trim.toInt, p(1), p(2), p(3).trim.toInt)).toDF()
// Vemos que tiene el archivo
dfFilms.show
//Registramos el dataframe para poder hacer consultas sobre él
dfFilms.registerTempTable("films")
//Realizamos una consulta con joins al datafame actos_films para saber cuantos actores trabajaron en los films.
val dfFilmsCountActors = spark.sql("select f.title, count(af.film_id) from films f join actors_films af on (f.film_id=af.film_id) group by f.title")
//Registramos el dataframe como una tabla temporal
dfFilmsCountActors.registerTempTable("films_Count_Actors")
//Hacemos una consulta sobre esa tabla con filtro.
spark.sql("select * from films_Count_Actors fca where fca.title like '%SANTA%' ").show
//Por ultimo exportamos este resultado.
dfFilmsCountActors.coalesce(1).write.csv("/home/spark/filmsCount.csv")
Fin!
La idea era poder hacer un conjunto de acciones que nos sirvan como ejemplo de las cosas que se pueden hacer con Spark SQL. Espero que les sirva!!
