<jupyter_start><jupyter_text>Spark tutorialFrom https://github.com/databricks/Spark-The-Definitive-GuideAnother source containing useful examples: https://github.com/hopelessoptimism/data-scientists-guide-apache-spark Spark has several core abstractions: Datasets, DataFrames, SQL Tables, and Resilient Distributed Datasets (RDDs). These abstractions all represent distributed collections of data however they have di erent interfaces for working with that data. The easiest and most e icient are DataFrames, which are available in all languages.  TransformationsThere are two type of transformation in dataframes:- Narrow; perform an operation called pipelining on narrow dependencies,  this means that if we specify multiple filters on DataFrames  they’ll all be performed in-memory- Wide; will write the results to disk Lazy evaluationLazy evaluation means that Spark will wait until the very last moment toexecute the graph of computation instructions via a *plan* of transformations.This provides immense benefits to the end user because Spark can optimizethe entire data flow from end to end. ActionsTransformations allow us to build up our logical transformation plan.To trigger the computation, we run an action. Spark UIUsers can monitor the progress of their job through the Spark UI.The Spark UI is available on port 4040 of the driver node.<jupyter_code>// Dataframe
val myRange = spark.range(1000).toDF("number")
// "where" transformation
val divisBy2 = myRange.where("number % 2 = 0")
// "count" action
divisBy2.count()
val flightData2015 = spark
   .read
   .option("inferSchema", "true")
   .option("header", "true")
   .csv("spark-data/2015-summary.csv")
// Get the data frame schema
flightData2015.schema
// Take the first rows
flightData2015.take(3)<jupyter_output><empty_output><jupyter_text>Explain Physical planExplain plans are a bit arcane, but with a bit of practice it becomes second nature. Explain plans can be read from top to bottom, the top being the end result and the bottom being the source(s) of data.<jupyter_code>flightData2015.sort("count").explain()<jupyter_output>== Physical Plan ==
*(2) Sort [count#22 ASC NULLS FIRST], true, 0
+- Exchange rangepartitioning(count#22 ASC NULLS FIRST, 200)
   +- *(1) FileScan csv [DEST_COUNTRY_NAME#20,ORIGIN_COUNTRY_NAME#21,count#22] Batched: false, Format: CSV, Location: InMemoryFileIndex[file:/home/filippos/Documents/notebooks/fsquillace-notebooks/tutorials/2015-sum..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<DEST_COUNTRY_NAME:string,ORIGIN_COUNTRY_NAME:string,count:int><jupyter_text>Change the number of partitionsBy default, when we perform a shuffle Spark will output two hundred shuffle partitions. We will set this value to five in order to reduce the number of the output partitions from the shuffle from two hundred to five.<jupyter_code>spark.conf.set("spark.sql.shuffle.partitions", "5")

flightData2015.sort("count").take(2)<jupyter_output><empty_output><jupyter_text>More details about partitionsMore info here:- https://jaceklaskowski.gitbooks.io/mastering-apache-spark/spark-rdd-partitions.html- https://spark.apache.org/docs/latest/tuning.htmlSpark splits data into partitions and executes computations on the partitions in parallel.<jupyter_code>val x = (1 to 10).toList
val numbersDf = x.toDF("number")
numbersDf.rdd.partitions.size 
// Each partition is a separate CSV file when you write a DataFrame to disc.

numbersDf.write.csv("./numbers")<jupyter_output><empty_output><jupyter_text>CoalescenceThe coalesce method reduces the number of partitions in a DataFrame. It does this by moving data between nodes from some partitions to existing partitions.**NOTE**:- Coalescence cannot increase the number of partitions.- Coalesce does not equally distribute the data among partitions (repartition does it).<jupyter_code>val numbersDf2 = numbersDf.coalesce(2)

numbersDf2.rdd.partitions.size
// The coalesce does not increase the number of partitions!
val numbersDf3 = numbersDf.coalesce(numbersDf.rdd.partitions.size + 5)

numbersDf3.rdd.partitions.size<jupyter_output><empty_output><jupyter_text>RepartitionThe repartition method can be used to either increase or decrease the number of partitions in a DataFrame.**NOTE**: Repartition does a full data shuffle and equally distribute the data among partitions. It does not minimize the number of data movement.<jupyter_code>val numbersDf4 = numbersDf.repartition(numbersDf.rdd.partitions.size + 5)

numbersDf3.rdd.partitions.size<jupyter_output><empty_output><jupyter_text>Index DataFrame by a given columnWhen partitioning by a column, Spark will create a minimum of 200 partitions by default.Partitioning by a column is similar to indexing a column in a relational database. Each partition group by the column's values.<jupyter_code>val people = List(
  (10, "blue"),
  (13, "red"),
  (15, "blue"),
  (99, "red"),
  (67, "blue")
)
val peopleDf = people.toDF("age", "color")

val colorDf = peopleDf.repartition($"color")

colorDf.rdd.partitions.size<jupyter_output><empty_output><jupyter_text>What number of partition to choose?`````` I/O bound taskIf you’re writing the data out to a file system, you can choose a partition size that will create reasonable sized files (100MB). Spark will optimize the number of partitions based on the number of clusters when the data is read.  CPU bound taskYou can determine the number of partitions by multiplying the number of CPUs in the cluster by 2, 3, or 4 (see more [here](https://stackoverflow.com/questions/35800795/number-of-partitions-in-rdd-and-performance-in-spark/3580440735804407) and [here](http://spark.apache.org/docs/latest/tuning.htmllevel-of-parallelism)).<jupyter_code>val numberCPUsPerCluster = sc.defaultParallelism
val numberPartitions = numberCPUsPerCluster * 4<jupyter_output><empty_output><jupyter_text>Spark SQLSpark SQL allows you as a user to register any DataFrame as a table or view (a temporary table) and query it using pure SQL.<jupyter_code>flightData2015.createOrReplaceTempView("flight_data_2015")
val sqlWay = spark.sql("""
    SELECT DEST_COUNTRY_NAME, count(1)
    FROM flight_data_2015
    GROUP BY DEST_COUNTRY_NAME
""")

val dataFrameWay = flightData2015
   .groupBy("DEST_COUNTRY_NAME")
   .count()

sqlWay.explain
dataFrameWay.explain
// In SQL:

spark.sql("SELECT max(count) FROM flight_data_2015").take(1)
// In Scala

import org.apache.spark.sql.functions

flightData2015.select(max("count")).take(1)
spark.sql("""
    SELECT DEST_COUNTRY_NAME, sum(count) AS destination_total
    FROM flight_data_2015
    GROUP BY DEST_COUNTRY_NAME
    ORDER BY destination_total DESC
    LIMIT 5
""").show()
import org.apache.spark.sql.functions.sum
import org.apache.spark.sql.functions.desc

flightData2015
    .groupBy("DEST_COUNTRY_NAME")
    .sum("count")
    .withColumnRenamed("sum(count)", "destination_total")
    .orderBy(desc("destination_total"))
    .limit(5)
    .show()<jupyter_output>+-----------------+-----------------+
|DEST_COUNTRY_NAME|destination_total|
+-----------------+-----------------+
|    United States|           411352|
|           Canada|             8399|
|           Mexico|             7140|
|   United Kingdom|             2025|
|            Japan|             1548|
+-----------------+-----------------+<jupyter_text>Datasets: Type-safe Structured APIs<jupyter_code>// A Scala case class (similar to a struct) that will automatically
// be mapped into a structured data table in Spark
case class Flight(DEST_COUNTRY_NAME: String, ORIGIN_COUNTRY_NAME: String, count: BigInt)

// The following import is required:
// https://stackoverflow.com/questions/38664972/why-is-unable-to-find-encoder-for-type-stored-in-a-dataset-when-creating-a-dat
import spark.implicits._
val flightsDF = spark.read.parquet("spark-data/2010-summary.parquet/")
val flights = flightsDF.as[Flight]
flights
  // This does not work: https://stackoverflow.com/a/42780897
//   .filter(flight_row => flight_row.ORIGIN_COUNTRY_NAME != "Canada")
  .filter(col("ORIGIN_COUNTRY_NAME") =!= "Canada")
  .show(5)<jupyter_output>+-----------------+-------------------+-----+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|
+-----------------+-------------------+-----+
|    United States|            Romania|    1|
|    United States|            Ireland|  264|
|    United States|              India|   69|
|            Egypt|      United States|   24|
|Equatorial Guinea|      United States|    1|
+-----------------+-------------------+-----+
only showing top 5 rows<jupyter_text>Structured streaming<jupyter_code>val staticDataFrame = spark.read.format("csv")
   .option("header", "true")
   .option("inferSchema", "true")
   .load("spark-data/retail-data/by-day/*.csv")

staticDataFrame.createOrReplaceTempView("retail_data")
val staticSchema = staticDataFrame.schema
staticDataFrame.show(5)<jupyter_output>+---------+---------+--------------------+--------+-------------------+---------+----------+--------------+
|InvoiceNo|StockCode|         Description|Quantity|        InvoiceDate|UnitPrice|CustomerID|       Country|
+---------+---------+--------------------+--------+-------------------+---------+----------+--------------+
|   580538|    23084|  RABBIT NIGHT LIGHT|      48|2011-12-05 08:38:00|     1.79|   14075.0|United Kingdom|
|   580538|    23077| DOUGHNUT LIP GLOSS |      20|2011-12-05 08:38:00|     1.25|   14075.0|United Kingdom|
|   580538|    22906|12 MESSAGE CARDS ...|      24|2011-12-05 08:38:00|     1.65|   14075.0|United Kingdom|
|   580538|    21914|BLUE HARMONICA IN...|      24|2011-12-05 08:38:00|     1.25|   14075.0|United Kingdom|
|   580538|    22467|   GUMBALL COAT RACK|       6|2011-12-05 08:38:00|     2.55|   14075.0|United Kingdom|
+---------+---------+--------------------+--------+-------------------+---------+----------+--------------+
only showing top 5 rows<jupyter_text>Simple query based on time window intervalIn this example we’ll take a look at the largest sale hours where a given customer (identified by CustomerId) makes a large purchase. For example, let’s add a total cost column and see on what days a customer spent the most.<jupyter_code>import org.apache.spark.sql.functions.{window, column, desc, col}
staticDataFrame
   .selectExpr(
   "CustomerId",
   "(UnitPrice * Quantity) as total_cost",
   "InvoiceDate")
   .groupBy(
     col("CustomerId"), window(col("InvoiceDate"), "1 day"))
   .sum("total_cost")
   .show(5, false)<jupyter_output>+----------+------------------------------------------+-----------------+
|CustomerId|window                                    |sum(total_cost)  |
+----------+------------------------------------------+-----------------+
|16057.0   |[2011-12-05 00:00:00, 2011-12-06 00:00:00]|-37.6            |
|14126.0   |[2011-11-29 00:00:00, 2011-11-30 00:00:00]|643.6300000000001|
|13500.0   |[2011-11-16 00:00:00, 2011-11-17 00:00:00]|497.9700000000001|
|17160.0   |[2011-11-08 00:00:00, 2011-11-09 00:00:00]|516.8499999999999|
|15608.0   |[2011-11-11 00:00:00, 2011-11-12 00:00:00]|122.4            |
+----------+------------------------------------------+-----------------+
only showing top 5 rows<jupyter_text>Streaming<jupyter_code>val streamingDataFrame = spark.readStream
    .schema(staticSchema)
    .option("maxFilesPerTrigger", 1) // How many files to read at once 
    .format("csv")
    .option("header", "true")
    .load("spark-data/retail-data/by-day/*.csv")
// Now we can see the DataFrame is streaming.
streamingDataFrame.isStreaming
val purchaseByCustomerPerHour = streamingDataFrame
   .selectExpr(
       "CustomerId",
       "(UnitPrice * Quantity) as total_cost",
       "InvoiceDate")
   .groupBy(
     col("CustomerId"), window(col("InvoiceDate"), "1 day"))
   .sum("total_cost")<jupyter_output><empty_output><jupyter_text>NoteBefore kicking off the stream, we will set a small optimization that will allow this to run better on a single machine. This simply limits the number of output partitions a er a shuffle<jupyter_code>spark.conf.set("spark.sql.shu le.partitions", "5")
purchaseByCustomerPerHour.writeStream
   .format("memory") // memory = store in-memory table
   .queryName("customer_purchases") // counts = name of the in-memory table
   .outputMode("complete") // complete = all the counts should be in the table
   .start()
// Another option you can use is to just simply write the results out to the console.
// This might take a while:
// purchaseByCustomerPerHour.writeStream
//   .format("console")
//   .queryName("customer_purchases_2")
//   .outputMode("complete")
// .start()<jupyter_output><empty_output><jupyter_text>Once we start the stream, we can run queries against the stream to debug what our result will look like if we were to write this out to a production sink.<jupyter_code>spark.sql("""
   SELECT *
   FROM customer_purchases
   ORDER BY `sum(total_cost)` DESC
   """)
   .show(5, false)<jupyter_output>+----------+------------------------------------------+------------------+
|CustomerId|window                                    |sum(total_cost)   |
+----------+------------------------------------------+------------------+
|null      |[2011-03-29 00:00:00, 2011-03-30 00:00:00]|33521.39999999998 |
|null      |[2011-12-08 00:00:00, 2011-12-09 00:00:00]|31975.590000000007|
|18102.0   |[2010-12-07 00:00:00, 2010-12-08 00:00:00]|25920.37          |
|null      |[2010-12-10 00:00:00, 2010-12-11 00:00:00]|25399.560000000012|
|null      |[2010-12-17 00:00:00, 2010-12-18 00:00:00]|25371.769999999768|
+----------+------------------------------------------+------------------+
only showing top 5 rows<jupyter_text>Machine learning<jupyter_code>staticDataFrame.printSchema()<jupyter_output>root
 |-- InvoiceNo: string (nullable = true)
 |-- StockCode: string (nullable = true)
 |-- Description: string (nullable = true)
 |-- Quantity: integer (nullable = true)
 |-- InvoiceDate: timestamp (nullable = true)
 |-- UnitPrice: double (nullable = true)
 |-- CustomerID: double (nullable = true)
 |-- Country: string (nullable = true)<jupyter_text>Data transformation into numericalMachine learning algorithms in MLlib require data to be represented as numerical values. Our current data is represented by a variety of di erent types including timestamps, integers, and strings. Therefore we need to transform this data into some numerical representation. In this instance, we will use several DataFrame transformations to manipulate our date data.<jupyter_code>import org.apache.spark.sql.functions.date_format

val preppedDataFrame = staticDataFrame
   .na.fill(0)  // convert n/a value into 0
   .withColumn("day_of_week", date_format($"InvoiceDate", "EEEE")) // Add new field based on InvoiceDate.
   .coalesce(5) // avoids data movement, but only if you are decreasing the number of RDD partitions.
preppedDataFrame.printSchema()
preppedDataFrame.show(5)<jupyter_output>+---------+---------+--------------------+--------+-------------------+---------+----------+--------------+-----------+
|InvoiceNo|StockCode|         Description|Quantity|        InvoiceDate|UnitPrice|CustomerID|       Country|day_of_week|
+---------+---------+--------------------+--------+-------------------+---------+----------+--------------+-----------+
|   580538|    23084|  RABBIT NIGHT LIGHT|      48|2011-12-05 08:38:00|     1.79|   14075.0|United Kingdom|     Monday|
|   580538|    23077| DOUGHNUT LIP GLOSS |      20|2011-12-05 08:38:00|     1.25|   14075.0|United Kingdom|     Monday|
|   580538|    22906|12 MESSAGE CARDS ...|      24|2011-12-05 08:38:00|     1.65|   14075.0|United Kingdom|     Monday|
|   580538|    21914|BLUE HARMONICA IN...|      24|2011-12-05 08:38:00|     1.25|   14075.0|United Kingdom|     Monday|
|   580538|    22467|   GUMBALL COAT RACK|       6|2011-12-05 08:38:00|     2.55|   14075.0|United Kingdom|     Monday|
+---------+---------+-------------------[...]<jupyter_text>Dataset split: training/test sets<jupyter_code>val trainDataFrame = preppedDataFrame
   .where("InvoiceDate < '2011-07-01'")
val testDataFrame = preppedDataFrame
   .where("InvoiceDate >= '2011-07-01'")

print(trainDataFrame.count())
print(testDataFrame.count())
import org.apache.spark.ml.feature.StringIndexer
val indexer = new StringIndexer()
   .setInputCol("day_of_week")
   .setOutputCol("day_of_week_index")<jupyter_output><empty_output><jupyter_text>This will turn our days of weeks into corresponding numerical values. For example, Spark may represent Saturday as 6 and Monday as 1. However with this numbering scheme, we are implicitly stating that Saturday is greater than Monday (by pure numerical values). This is obviously incorrect. Therefore we need to use a OneHotEncoder to encode each of these values as their own column. These boolean flags state whether that day of week is the relevant day of the week.<jupyter_code>import org.apache.spark.ml.feature.OneHotEncoder

val encoder = new OneHotEncoder()
   .setInputCol("day_of_week_index")
   .setOutputCol("day_of_week_encoded")<jupyter_output><empty_output><jupyter_text>All machine learning algorithms in Spark take as input a Vector type, which must be a set of numerical values.<jupyter_code>import org.apache.spark.ml.feature.VectorAssembler

val vectorAssembler = new VectorAssembler()
  .setInputCols(Array("UnitPrice", "Quantity", "day_of_week_encoded"))
  .setOutputCol("features")
import org.apache.spark.ml.Pipeline

val transformationPipeline = new Pipeline()
   .setStages(Array(indexer, encoder, vectorAssembler))<jupyter_output><empty_output><jupyter_text>`StringIndexer` needs to know how many unique values there are to be index. In this way it will map Monday to 1, Tuesday to 2, etc.<jupyter_code>val fittedPipeline = transformationPipeline.fit(trainDataFrame)<jupyter_output><empty_output><jupyter_text>Once we fit the training data, we are now create to take that fitted pipeline and use it to transform all of our data in a consistent and repeatable way.<jupyter_code>val transformedTraining = fittedPipeline.transform(trainDataFrame)
transformedTraining.cache()
transformedTraining.show(5)<jupyter_output>+---------+---------+--------------------+--------+-------------------+---------+----------+--------------+-----------+-----------------+-------------------+--------------------+
|InvoiceNo|StockCode|         Description|Quantity|        InvoiceDate|UnitPrice|CustomerID|       Country|day_of_week|day_of_week_index|day_of_week_encoded|            features|
+---------+---------+--------------------+--------+-------------------+---------+----------+--------------+-----------+-----------------+-------------------+--------------------+
|   537226|    22811|SET OF 6 T-LIGHTS...|       6|2010-12-06 08:34:00|     2.95|   15987.0|United Kingdom|     Monday|              2.0|      (5,[2],[1.0])|(7,[0,1,4],[2.95,...|
|   537226|    21713|CITRONELLA CANDLE...|       8|2010-12-06 08:34:00|      2.1|   15987.0|United Kingdom|     Monday|              2.0|      (5,[2],[1.0])|(7,[0,1,4],[2.1,8...|
|   537226|    22927|GREEN GIANT GARDE...|       2|2010-12-06 08:34:00|     5.95|   15987.0|United Kingdo[...]<jupyter_text>ML modelsIn Spark, training machine learning models is a two phase process. First we initialize an untrained model, then we train it. There are always two types for every algorithm in MLlib’s DataFrame API. They following the naming pattern of `Algorithm`, for the untrained version, and `AlgorithmModel` for the trained version. In our case, this is `KMeans` and then `KMeansModel`.<jupyter_code>import org.apache.spark.ml.clustering.KMeans
val kmeans = new KMeans()
   .setK(20)
   .setSeed(1L)
val kmModel = kmeans.fit(transformedTraining)
kmModel.computeCost(transformedTraining)
val transformedTest = fittedPipeline.transform(testDataFrame)
kmModel.computeCost(transformedTest)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.18057448634459083, 'fraction_numerical': 0.059669847039224594, 'mean_word_length': 3.0461601307189543, 'pattern_counts': {'":': 0, '<': 86, '<?xml version=': 0, '>': 87, 'https://': 7, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 7, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '23', 'dir': 'jupyter-structured-clean-dedup', 'id': '130864', 'license': 'MIT', 'n_tokens_mistral': 6817, 'n_tokens_neox': 5872, 'n_words': 1659, 'path': 'notebooks/learning/spark/spark.ipynb', 'repo_name': 'fsquillace/notebooks', 'size': '19809'}