... Certificazione Apache Spark Developer con Scala – Parte 1

spark.version

res1: String = 2.4.4

spark.conf.set("spark.sql.shuffle.partitions", 6)
spark.conf.set("spark.executor.memory", "6g")

print(spark.conf.get("spark.sql.shuffle.partitions") + ", " + spark.conf.get("spark.executor.memory"))

6, 6g

spark.conf.getAll.foreach(println)

(zeppelin.pyspark.python,C:\Users\home\Anaconda3\envs\conda2020\python)
(spark.driver.host,10.0.75.1)
(zeppelin.dep.localrepo,local−repo)
(zeppelin.spark.sql.stacktrace,false)
(spark.driver.port,51622)
(master,local[15])
(spark.repl.class.uri,spark://10.0.75.1:51622/classes)
(zeppelin.spark.useHiveContext,true)
(spark.repl.class.outputDir,C:\Users\home\AppData\Local\Temp\spark6251788337677085436)
(zeppelin.spark.sql.interpolation,false)
(zeppelin.spark.importImplicit,true)
(zeppelin.interpreter.output.limit,102400)
(spark.app.name,Zeppelin)
(zeppelin.R.cmd,R)
(zeppelin.spark.maxResult,1000)
(zeppelin.pyspark.useIPython,true)
(zeppelin.spark.concurrentSQL,false)
(zeppelin.spark.enableSupportedVersionCheck,true)
(zeppelin.spark.printREPLOutput,true)
(zeppelin.dep.additionalRemoteRepository,spark−packages,http://dl.bintray.com/spark−packages/maven,false;)
(org.apache.spark.storage.BlockManager,DEBUG)
(spark.executor.id,driver)
(log4j.logger.org.apache.spark.storage.BlockManager,TRACE)
(zeppelin.spark.useNew,true)
(spark.useHiveContext,true)
(spark.master,local)
(zeppelin.R.image.width,100%)
(zeppelin.spark.ui.hidden,false)
(zeppelin.interpreter.localRepo,C:\zeppelin−0.8.2−bin−all/local−repo/spark)
(spark.executor.memory,6g)
(spark.driver.allowMultipleContexts,false)
(zeppelin.R.render.options,out.format = 'html', comment = NA, echo = FALSE, results = 'asis', message = F, warning = F, fig.retina = 2)
(zeppelin.interpreter.max.poolsize,10)
(spark.app.id,local−1578911285841)
(zeppelin.R.knitr,true)
(spark.sql.shuffle.partitions,6)

sc.stop()

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
val conf = new SparkConf().setMaster("local[*]").setAppName("provaApp")
val sc = new SparkContext(conf)

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
conf: org.apache.spark.SparkConf = org.apache.spark.SparkConf@13115401
sc: org.apache.spark.SparkContext = org.apache.spark.SparkContext@5a7f8768

conf.toDebugString

res4: String =
spark.app.name=provaApp
spark.driver.extraClassPath=C:\Users\home\Downloads\jar_files\mongo−spark−connector_2.12−2.4.1.jar
spark.driver.extraJavaOptions= −Dfile.encoding=UTF−8 −Dzeppelin.log.file='C:\zeppelin−0.8.2−bin−all\logs\zeppelin−interpreter−spark−home−DESKTOP−46GRV54.log'
spark.driver.memory=8g
spark.executor.extraClassPath=C:\Users\home\Downloads\jar_files\mongo−spark−connector_2.12−2.4.1.jar
spark.executor.memory=8g
spark.jars=file:///C:/zeppelin−0.8.2−bin−all/interpreter/spark/spark−interpreter−0.8.2.jar,file:/C:/zeppelin−0.8.2−bin−all/interpreter/spark/spark−interpreter−0.8.2.jar
spark.master=local[*]
spark.repl.local.jars=file:///C:/zeppelin−0.8.2−bin−all/interpreter/spark/spark−interpreter−0.8.2.jar
spark.submit.deployMode=client

spark.stop()

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

val conf = new SparkConf().setMaster("local[*]").set("spark.driver.cores", "1").set("spark.executor.memory","4G")

val spark = SparkSession.builder.config(conf=conf).getOrCreate()

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
conf: org.apache.spark.SparkConf = org.apache.spark.SparkConf@650f29a4
spark: org.apache.spark.sql.SparkSession = org.apache.spark.sql.SparkSession@7db780c

import spark.implicits._

val lista_numeri = List(1,2,3,4,5,6,7)
val list_df = lista_numeri.toDF()

import spark.implicits._
lista_numeri: List[Int] = List(1, 2, 3, 4, 5, 6, 7)
list_df: org.apache.spark.sql.DataFrame = [value: int]

StructType( 
    List ( StructField("name1", TypeName1), StructField("name2", TypeName2) ...) 
)

import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{StructField, StructType, IntegerType, StringType}
import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder.master("local[15]").appName("test").getOrCreate()

//creiamo un RDD da una sequenza
val someData = spark.sparkContext.parallelize( Seq(
         Row(10230, "Paolo"),
         Row(12025, "Pippo"),
         Row(25876, "Pedro")
         )
        )

//definiamo lo schema
val someSchema = StructType ( List(
         StructField("Identificativo", IntegerType, true),
         StructField("Nome", StringType, true)
         )
       )

//dobbiamo dare come parametri un RDD (creato con sc.parallelize) e lo schema
val someDF = spark.createDataFrame( someData, someSchema )

someDF.show()

+−−−−−−−−−−−−−−+−−−−−+
|Identificativo| Nome|
+−−−−−−−−−−−−−−+−−−−−+
|         10230|Paolo|
|         12025|Pippo|
|         25876|Pedro|
+−−−−−−−−−−−−−−+−−−−−+

import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{StructField, StructType, IntegerType, StringType}
someData: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = ParallelCollectionRDD[0] at parallelize at <console>:37
someSchema: org.apache.spark.sql.types.StructType = StructType(StructField(Identificativo,IntegerType,true), StructField(Nome,StringType,true))
someDF: org.apache.spark.sql.DataFrame = [Identificativo: int, Nome: string]

import org.apache.spark.sql.types.StructType

val schema = new StructType()
        .add($"Identificativo".long.copy(nullable = false))
        .add($"Nome".string)
        .add($"Cognome".string)

schema.printTreeString

import org.apache.spark.sql.DataFrameReader

val r: DataFrameReader = spark.read.schema(schema)

root
 |−− Identificativo: long (nullable = false)
 |−− Nome: string (nullable = true)
 |−− Cognome: string (nullable = true)

import org.apache.spark.sql.types.StructType
schema: org.apache.spark.sql.types.StructType = StructType(StructField(Identificativo,LongType,false), StructField(Nome,StringType,true), StructField(Cognome,StringType,true))
import org.apache.spark.sql.DataFrameReader
r: org.apache.spark.sql.DataFrameReader = org.apache.spark.sql.DataFrameReader@3e061032

val data_from_csv = r.csv("exampl1.csv")

//oppure

val data_from_load = r.load("exampl1.csv")

data_from_csv: org.apache.spark.sql.DataFrame = [Identificativo: bigint, Nome: string ... 1 more field]
data_from_load: org.apache.spark.sql.DataFrame = [Identificativo: bigint, Nome: string ... 1 more field]

val df = spark.createDataFrame( sc.parallelize( Map(("x", 24), ("y", 25), ("z", 26)).toSeq ) )

df.withColumnRenamed("_1", "nome").withColumnRenamed("_2", "età").show()

+−−−−+−−−+
|nome|età|
+−−−−+−−−+
|   x| 24|
|   y| 25|
|   z| 26|
+−−−−+−−−+

df: org.apache.spark.sql.DataFrame = [_1: string, _2: int]

import spark.implicits._

Map(("x", 24), ("y", 25), ("z", 26)).toSeq.toDF

import spark.implicits._
res8: org.apache.spark.sql.DataFrame = [_1: string, _2: int]

sc.parallelize(1 to 10).toDF()

res16: org.apache.spark.sql.DataFrame = [value: int]

import org.apache.spark.sql.DataFrame

val ar = 1 to 10
val df: DataFrame = ar.toDF()

import org.apache.spark.sql.DataFrame
ar: scala.collection.immutable.Range.Inclusive = Range(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
df: org.apache.spark.sql.DataFrame = [value: int]

spark.range(1, 1000).show(5)

+−−−+
| id|
+−−−+
|  1|
|  2|
|  3|
|  4|
|  5|
+−−−+
only showing top 5 rows

spark.range(1,100).toDF("ID").show(5)

+−−−+
| ID|
+−−−+
|  1|
|  2|
|  3|
|  4|
|  5|
+−−−+
only showing top 5 rows

import org.apache.spark.sql.DataFrameReader

val r: DataFrameReader = spark.read

r.csv("exampl1.csv")

import org.apache.spark.sql.DataFrameReader
r: org.apache.spark.sql.DataFrameReader = org.apache.spark.sql.DataFrameReader@3f709ba
res24: org.apache.spark.sql.DataFrame = [_c0: string, _c1: string]

val df = spark.read.csv("exampl1.csv")

df: org.apache.spark.sql.DataFrame = [_c0: string, _c1: string]

val csvLine = "0,Warsaw,Poland"

import org.apache.spark.sql.Dataset
val cities: Dataset[String] = Seq(csvLine).toDS

cities.show

// Define schema explicitly (as below)
// or
// option("header", true) + option("inferSchema", true)
import org.apache.spark.sql.types.StructType
val schema = new StructType()
  .add($"id".long.copy(nullable = false))
  .add($"city".string)
  .add($"country".string)

schema.printTreeString

import org.apache.spark.sql.DataFrame

val citiesDF: DataFrame = spark
  .read
  .schema(schema)
  .csv(cities)
 
citiesDF.show

+−−−−−−−−−−−−−−−+
|          value|
+−−−−−−−−−−−−−−−+
|0,Warsaw,Poland|
+−−−−−−−−−−−−−−−+

root
 |−− id: long (nullable = false)
 |−− city: string (nullable = true)
 |−− country: string (nullable = true)

+−−−+−−−−−−+−−−−−−−+
| id|  city|country|
+−−−+−−−−−−+−−−−−−−+
|  0|Warsaw| Poland|
+−−−+−−−−−−+−−−−−−−+

csvLine: String = 0,Warsaw,Poland
import org.apache.spark.sql.Dataset
cities: org.apache.spark.sql.Dataset[String] = [value: string]
import org.apache.spark.sql.types.StructType
schema: org.apache.spark.sql.types.StructType = StructType(StructField(id,LongType,false), StructField(city,StringType,true), StructField(country,StringType,true))
import org.apache.spark.sql.DataFrame
citiesDF: org.apache.spark.sql.DataFrame = [id: bigint, city: string ... 1 more field]

//importiamo udf
import org.apache.spark.sql.functions.udf

val power3udf = udf( (x:Double) => ( x*x*x ) )

val df = spark.sparkContext.parallelize(0 to 10 by 2).toDF()

df.show()

df.select(power3udf( $"value" )).show()

+−−−−−+
|value|
+−−−−−+
|    0|
|    2|
|    4|
|    6|
|    8|
|   10|
+−−−−−+

+−−−−−−−−−−+
|UDF(value)|
+−−−−−−−−−−+
|       0.0|
|       8.0|
|      64.0|
|     216.0|
|     512.0|
|    1000.0|
+−−−−−−−−−−+

import org.apache.spark.sql.functions.udf
power3udf: org.apache.spark.sql.expressions.UserDefinedFunction = UserDefinedFunction(<function1>,DoubleType,Some(List(DoubleType)))
df: org.apache.spark.sql.DataFrame = [value: int]

import org.apache.spark.sql.types.StringType

def lowerRemoveAllWhitespace(s: String): String = {
 s.toLowerCase().replaceAll("\\s", "")
}

val lowerRemoveAllWhitespaceUDF = udf[String, String](lowerRemoveAllWhitespace)

val sourceDF = List(
 ("  HI THERE     "),
 (" GivE mE PresenTS     ")).toDF()

sourceDF.show()

sourceDF.select(
 lowerRemoveAllWhitespaceUDF(col("value")).as("clean_value")
).show()

+−−−−−−−−−−−−−−−−−−−−+
|               value|
+−−−−−−−−−−−−−−−−−−−−+
|       HI THERE     |
| GivE mE PresenTS...|
+−−−−−−−−−−−−−−−−−−−−+

+−−−−−−−−−−−−−−+
|   clean_value|
+−−−−−−−−−−−−−−+
|       hithere|
|givemepresents|
+−−−−−−−−−−−−−−+

import org.apache.spark.sql.types.StringType
lowerRemoveAllWhitespace: (s: String)String
lowerRemoveAllWhitespaceUDF: org.apache.spark.sql.expressions.UserDefinedFunction = UserDefinedFunction(<function1>,StringType,Some(List(StringType)))
sourceDF: org.apache.spark.sql.DataFrame = [value: string]

//creiamo un dataframe
val df = (1 to 100 by 3).toDF("num")
df.show(5)

//registrimo la funzione in SQL
spark.udf.register("power2", (x: Double) => x*x)

//adesso che la funzione è registrata posso usarla in una query sql, nel formato Scala
df.selectExpr("power2(num)").show(5)

//posso usare la funzione anche nel formato SQL
//prima creiamo una view temporanea
df.createOrReplaceTempView("udfExampleSQL3")
//e poi la query
spark.sql("SELECT power2(num) AS p2 FROM udfExampleSQL3").show(5)

+−−−+
|num|
+−−−+
|  1|
|  4|
|  7|
| 10|
| 13|
+−−−+
only showing top 5 rows

+−−−−−−−−−−−−−−−−−−−−−−−−−−−−−−−+
|UDF:power2(cast(num as double))|
+−−−−−−−−−−−−−−−−−−−−−−−−−−−−−−−+
|                            1.0|
|                           16.0|
|                           49.0|
|                          100.0|
|                          169.0|
+−−−−−−−−−−−−−−−−−−−−−−−−−−−−−−−+
only showing top 5 rows

+−−−−−+
|   p2|
+−−−−−+
|  1.0|
| 16.0|
| 49.0|
|100.0|
|169.0|
+−−−−−+
only showing top 5 rows

df: org.apache.spark.sql.DataFrame = [num: int]

spark.read.format("csv")
    .option("mode", "FAILFAST")
    .option("inferSchema", "true")
    .option("path", "path/to/file(s)")
    .schema(someSchema)
    .load()

val dataFrame = spark.read.json("example.json")

val dataFrame = spark.read.csv("example.csv") 

val dataFrame = spark.read.parquet("example.parquet")

val dataFrame = spark.read.text("file.txt")

val dataFrame = spark.read.orc("file.orc")

val dataFrame = spark.read.jdbc(url,"person",properties)

spark.read.format("json").load("example.json")

spark.read.format("csv").load("example.csv")

spark.read.format("parquet").load("example.parquet")

spark.read.format("text").load("file.txt")

val df = spark.read.option("header", "true").option("inferSchema", "true").csv("exampl1.csv")
//or 
val df2 = spark.read.options(Map(("header", "true"), ("inferSchema","true"))).csv("exampl1.csv")
//or
val df3 = spark.read.format("csv").option("header", "true").option("inferSchema", "true").option("mode", "FAILFAST").load("exampl1.csv")

df: org.apache.spark.sql.DataFrame = [eta: int, amici: int]
df2: org.apache.spark.sql.DataFrame = [eta: int, amici: int]
df3: org.apache.spark.sql.DataFrame = [eta: int, amici: int]

val df = spark.read.format("txt").option("sep","-").option("inferSchema","true").option("header","true").option("ignoreLeadingWhiteSpace","true").option("comment", "<!--").load(data_file)

Come già visto in precedenza, dobbiamo importare StructType and StructField. E costruire lo schema come nel codice sottostante.

import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType, FloatType}

val my_schema = StructType( List(
        StructField("nome", StringType),
        StructField("cognome", StringType, nullable=true),
        StructField("altezza", IntegerType, nullable=true), 
        StructField("peso", FloatType, nullable=true),
        StructField("età", IntegerType)
        )
       )

val df = spark.read.format("csv").schema(my_schema).option("mode","PERMISSIVE").load("exampl2.csv")
df.show()

+−−−−−−−−+−−−−−−−−−+−−−−−−−+−−−−+−−−+
|    nome|  cognome|altezza|peso|età|
+−−−−−−−−+−−−−−−−−−+−−−−−−−+−−−−+−−−+
|   Pippo|  Paolini|    167|72.3| 44|
|   Luigi|    Russo|    178|89.2|  5|
|Giovanna|    Rosso|    175|80.0| 44|
|Giuseppe|  Bianchi|    165|75.8|  9|
|  Amedeo|    Verdi|    167|56.2| 18|
|   Luisa|Valentino|    182|64.9| 15|
+−−−−−−−−+−−−−−−−−−+−−−−−−−+−−−−+−−−+

import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType, FloatType}
my_schema: org.apache.spark.sql.types.StructType = StructType(StructField(nome,StringType,true), StructField(cognome,StringType,true), StructField(altezza,IntegerType,true), StructField(peso,FloatType,true), StructField(età,IntegerType,true))
df: org.apache.spark.sql.DataFrame = [nome: string, cognome: string ... 3 more fields]

spark.read.schema("a INT, b STRING, c DOUBLE").csv("test.csv")

df.write.parquet(folder-path)

df.write.orc(folder-path)

df.write.json(folder-path)

df.write.csv(folder-path)

df.write.text(folder-path)

df.write.format("csv").mode("overwrite").option("sep", "\t").save("/my-tsv-file.tsv")

df.write.saveAsTable("mytable")

// df.write.json(folder-path)
// oppure 

val df = (1 to 10).toDF()

df.write.format("json").save("d:\\AnacondaProjects\\testSave")

df: org.apache.spark.sql.DataFrame = [value: int]

df.write.mode("append").csv("data.csv")

csvFile.write.format("csv").mode("overwrite").option("sep", "\t").save("my-tsv-file.tsv")

df.write.option("sep", ",")
     .option("quote", "U+005C")
     .option("escape", "U+005C")
     .option("charToEscapeQuoteEscaping", "\0")
     .option("escapeQuotes", "true")
     .option("quoteAll", false)
     ...

df.coalesce(1).write.csv("D:\\Anacondaprojects\\prova")
df.repartition(10).write.csv("D:\\Anacondaprojects\\prova2")

df.write.partitionBy("country").csv("./folder1")

Il bucketing è una tecnica che usa i buckets (contenitori) per definire il partizionamento dei dati e ridurre lo shuffle dei dati stessi.

Si veda la pagina Bucketing su The internal of Spark SQL.

Nella pagina linkata è mostrato come facendo il join di due dataframe salvati con il metodo .bucketBy() ci si risparmia lo shuffle nell’operazione di join. Si noti che avendo salvato con bucketBy i dati sono pre-shuffled.

Per eseguire il bucketing devo indicare quanti file salvare e su quale colonna di dati fare il bucketing/partizionamento.