anchor.diff:  { "name": "Bob", "public_key": "2800iBLw", "private_key": "jn1tCP/4" }
   ----
   
  -
  -Let's load the data.
  -
  -.Loading the `users.json` file as a _DataFrame_
  -[source,scala]
  -----
  -val dataframe = spark.read.json("users.json")
  -----
  -
  -Now, let's find out what was loaded.
  -
  -.Printing the content of our _DataFrame_
  +Let's load the data (listing 1.2).
  +
  +.Loading the `users.json` file as a DataFrame
  +[source,scala]
  +----
  +val dataframe = spark // <1>
  +  .read               // <2>
  +  .json("users.json") // <3>
  +----
  +<1> "spark" designates the SparkSession made available in your REPL, you can
  +create one yourself with `SparkSession.builder().getOrCreate()` assuming you
  +imported `org.apache.spark.sql.SparkSession`.
  +<2> We want to read stuff from a data source.
  +<3> This data source happens to be JSON and the file happens to be in the
  +working directory under the name `users.json`.
  +
  +Now, let's find out what was loaded (listing 1.3).
  +
  +.Printing the content of the DataFrame
   [source,scala]
   ----
   dataframe.show()
  -
  -// +-----+-----------+----------+
  -// | name|private_key|public_key|
  -// +-----+-----------+----------+
  -// |Alice|   656qoAVR|  MIIEogIB|
  -// |  Bob|   jn1tCP/4|  2800iBLw|
  -// +-----+-----------+----------+
  -----
  -
  -Next, let's project on the name column.
  -
  -.Projecting our _DataFrame_ on the name column
  -[source,scala]
  -----
  -dataframe.select("name").show()
  -
  -// +-----+
  -// | name|
  -// +-----+
  -// |Alice|
  -// |  Bob|
  -// +-----+
  -----
  -
  -As mentioned earlier, you can also query your _DataFrame_ using vanilla SQL:
  -
  -.Projecting our _DataFrame_ on the name column using SQL
  -[source,scala]
  -----
  -// We need to register the DataFrame as a SQL table (here as a temporary view)
  -dataframe.createOrReplaceTempView("users")
  -
  -spark.sql("SELECT name FROM users").show()
  -----
  +----
  +
  +This gives the following result (table 1.3):
  +
  +.Result of dataframe.show()
  +[options="header"]
  +|===
  +| name|private_key|public_key
  +|Alice|   656qoAVR|  MIIEogIB
  +|  Bob|   jn1tCP/4|  2800iBLw
  +|===
  +
  +You can find out what Spark knows about the schema of your data with
  +(listing 1.4):
  +
  +.Finding out what Spark knows about the schema of the DataFrame
  +[source,scala]
  +----
  +dataframe.printSchema()
  +----
  +
  +As assumed, there are 3 columns: "name", "private_key" and "public_key", all
  +strings.
  +
  +Next, in listing 1.5, let's project on the name column.
  +
  +.Projecting the DataFrame on the "name" column
  +[source,scala]
  +----
  +dataframe
  +  .select("name") // <1>
  +  .show()         // <2>
  +----
  +<1> We select the "name" column.
  +<2> We want Spark to show the result.
  +
  +Which gives (table 1.4):
  +
  +.Result of dataframe.select("name").show()
  +[options="header"]
  +|===
  +| name
  +|Alice
  +|  Bob
  +|===
  +
  +As mentioned, you can also query your DataFrame using SQL (listing 1.6):
  +
  +.Projecting our DataFrame on the name column using SQL
  +[source,scala]
  +----
  +dataframe.createOrReplaceTempView("users") // <1>
  +
  +spark
  +  .sql("SELECT name FROM users")           // <2>
  +  .show()
  +----
  +<1> We need to register the DataFrame as a SQL table (here as a temporary view).
  +<2> We input our raw SQL query through the `sql` method on our SparkSession.
   
   As you can see, we interact with columns just with strings with no knowledge
  -of the underlying type of the column we're dealing with. As a result, filtering
  -on a number for a string-typed column will work fine:
  -
  -.Filtering our _DataFrame_ using incompatible types
  -[source,scala]
  -----
  -dataframe.filter($"name" > 1).show()
  -
  -// +----+-----------+----------+
  -// |name|private_key|public_key|
  -// +----+-----------+----------+
  -// +----+-----------+----------+
  -----
  -
  -The result, an empty _DataFrame_, is kind of difficult to interpret, we would
  +of the underlying type of the column we're dealing with. As a result,
  +demonstrated in listing 1.7, filtering on a number for a string-typed column
  +will work fine:
  +
  +.Filtering the DataFrame using incompatible types
  +[source,scala]
  +----
  +dataframe
  +  .filter($"name" > 1) // <1>
  +  .show()
  +----
  +<1> We filter the rows where the value for the name column (which is a string)
  +is superior to 1 which doesn't make any sense!
  +
  +The result, an empty DataFrame, is kind of difficult to interpret, we would
   have expected a type error saying that you cannot compare a string and an
  -integer.
  +integer this is the consequence of the DataFrame API having weak typing. The
  +Dataset API remedy those shortcomings.
   
   === Dataset
   
  -The _Dataset_ API is similar to the _DataFrame_ API in the sense that you still
  +The Dataset API is similar to the DataFrame API in the sense that you still
   get the optimizations Catalyst provides but you also get the strong typing and
  -API familiarity, if you have a Scala background, from the _RDD_ API. It's
  +API familiarity, if you have a Scala background, from the RDD API. It's
   basically the best of both worlds.
   
  -Continuing with out previous example let's see the _Dataset_ API in action.
  -
  -We can turn our _DataFrame_ into a _Dataset_ by defining its model:
  -
  -.Turning a _DataFrame_ into a _Dataset_
  -[source,scala]
  -----
  -// Our user model, notice that the field names in our model must match the
  -// field names in the data
  -case class User(name: String, public_key: String, private_key: String)
  -val dataset = dataframe.as[User]
  -----
  -
  -You can still project using the _DataFrame_ API as we've shown earlier. But,
  -now you can use the familiar and typesafe collection API from Scala that the
  -_RDD_ API benefits from:
  -
  -.Projecting our _Dataset_ on the name using the _Dataset_ API
  -[source,scala]
  -----
  -dataset.map(user => user.name).show()
  -
  -// +-----+
  -// |value|
  -// +-----+
  -// |Alice|
  -// |  Bob|
  -// +-----+
  -----
  -
  -The added benefit is that you cannot do whatever you want with the different
  -types in your _Dataset_, this will not compile:
  +Continuing with the previous example, let's see the Dataset API in action.
  +
  +We can turn our DataFrame into a Dataset by defining its model (listing 1.8):
  +
  +.Turning a DataFrame into a Dataset
  +[source,scala]
  +----
  +case class User(name: String, public_key: String, private_key: String) // <1>
  +val dataset = dataframe.as[User]                                       // <2>
  +----
  +<1> Our user model, notice that the field names in our model must match the
  +field names in the data.
  +<2> We tell Spark that our DataFrame effectively contains User by turning it
  +into a Dataset through the `as` method which will map each record to the User
  +type.
  +
  +You can still project using the DataFrame API as we've shown earlier. But,
  +now you can use the familiar and typesafe collection API from Scala with
  +anonymous functions as shown in listing 1.9:
  +
  +.Projecting our Dataset on the name using the Dataset API
  +[source,scala]
  +----
  +dataset
  +  .map(user => user.name) // <1>
  +  .show()
  +----
  +<1> We map the user to its name since that's what is of interest to us.
  +
  +Which gives the same results than what was shown in table 1.4.
  +
  +The added benefit, compared to the DataFrame API, is that you cannot do whatever
  +you want with the different types in your Dataset, listing 1.10 will not
  +compile:
   
   .Trying to filter using incompatible types will result in a type error
   [source,scala]
  
positive.code: loop (REPL) which lets you experiment quickly and make quick and
  dirty prototypes. This is particularly important if you're in a "fail fast fail
  often" mindset.
  Another aspect, which goes hand in hand with the REPL, is a simple and
  functional API that lets you express distributed data processing operations as
  you would write local collection transformations using the Scala API.
  A lot of other components such as graph or stream processing are made available
  directly through Spark which saves you a lot of headaches when trying
  to use two existing but distinct frameworks together.
  
  Last but not least, Spark benefits from great integration with external data
  sources like Hive or Kafka which makes integrating Spark with your existing
  data infrastructure seamless.
  
  To give you an overview of what a Spark cluster looks like:
  
  .Overview of a Spark cluster with two nodes. The _SparkSession_ interacts with the cluster manager to obtain the required computation power. It also interacts with the executors to ditribute work.
  image::../images/cluster.png[]
  
  On the architecture side, each worker node will run its own JVM process called
  an _Executor_ in Spark parlance. Each _Executor_ will run its own set of tasks,
  in a multithreaded manner, which have been scheduled by the _SparkSession_. The
  _SparkSession_ is part of a driver program, another JVM, which can be
  run on a worker node or on your own computer. The _SparkSession_ is also
  responsible for interacting with the cluster manager which is responsible for
  allocating resources to your Spark application, more on that later.
  
  As opposed to other frameworks, when using Spark you can benefit from different
  subprojects which are integral components of Spark.
  
  .The different Spark subprojects: graph processing, stream processing, machine learning and the supported cluster managers
  image::../images/stack.png[]
  
negative.code: As its name implies, Spark Core contains the base set of functionalities you can
  expect from a distributed data processing framework: fault tolerance mechanisms,
  networking, serialization, interactions with distributed storage, executors,
  and so on.
  
  Resilient distributed datasets (RDDs), the low-level programming abstraction in
  Apache Spark, are also defined in the core. RDDs can be viewed
  as a strongly-typed distributed collection of items. One of the nice things
  when dealing with RDDs is that they define the same API you use when dealing
  with Scala collections, which reduces considerably the barrier of entry when
  using Spark by letting you do all kinds of transformations on distributed data
  as you would do them on non-distributed data.
  
  === Spark SQL
  
  Spark SQL is the module dedicated to structured data processing. As you might
  have guessed, it lets you interact with data using SQL; however, that's not all.
  Spark SQL also defines two concepts: _DataFrames_ and _Datasets_ which you can
  think of as distributed collections of data. I will expand on DataFrames and
  Datasets later in this chapter.
  
  The main benefit when using Spark SQL when compared with the traditional RDD API
  is that all the operations performed with Spark SQL go through an optimizer
  called _Catalyst_ (for more information about Catalyst go to
  http://people.csail.mit.edu/matei/papers/2015/sigmod_spark_sql.pdf). For
  example, with smart data sources like Parquet, when using a filter, data is
  filtered directly in the source and won't be retrieved into Spark. This is
  called a _filter pushdown_.
  
  As a result, no matter which language (Scala, Java, Python, R) or API
  (Dataset, DataFrame, SQL) you use, everything goes through the same optimizer
  and result in the same performance. This is different from the RDD API where you
  had the best performance using a language running on the JVM (Scala or Java)
  since you didn't have to deal with the overhead introduced by accessing Java
  objects in a JVM from a Python interpreter.
  
  === Spark Streaming
  
  Spark Streaming handles the real-time processing of data streams which can come
  from many message queue like systems like Apache Kafka, Amazon Kinesis or
  Apache Flume.
  
  The main programming abstraction in Spark Streaming is the discretized stream
  (_DStream_) which is an infinite stream of RDDs. Each RDD corresponds to the
  data received during a small timeframe which usually ranges from a couple of
  hundred of milliseconds to a couple of seconds.
  
  Because a DStream is just an infinite collection of RDDs, the same API is
  exposed, which makes turning a batch into a streaming application really easy,
  assuming you've already written the batch application, since a lot of the code
  will be reused and the core logic will be shared between the two.
  
  Suppose you are maintaning a few web services deployed on many different
  machines. A canonical example for streaming applications would be to
  analyze the logs of those web services and find out how many Internal Server
  Error people are hitting every hour.
  
  In addition to the DStream API, there is also a new one built on Spark SQL
  called Structured Streaming which is destined to be a replacement for Spark
  Streaming. In this case, the stream is actually represented as an unbounded
  DataFrame. I won't go into much detail about it since, at the time of writing,
  Structured Streaming is still in alpha release.
  
  As a result, there are two APIs at the moment to deal with real-time data
  processing: Spark Streaming and Structured Streaming.
  
  === GraphX
  
  GraphX is the subproject focused on graph processing. It relies on and extends
  the concept of RDD to build directed graphs with properties associated to
  each vertex and edge. A few algorithms are supported out of the box, such as
  PageRank, connected components or triangle counting. If you want to find more, I
  invite you to read the GraphX guide at
  http://spark.apache.org/docs/latest/graphx-programming-guide.html.
  
  === MLlib
  
  MLlib is the library dedicated to distributed machine learning in Spark.
  
  Much like Spark Streaming, MLlib is split in two based on which API it relies
  on. On one hand, the historical RDD-based API which is in maintenance mode. On
  the other is the DataFrame/Dataset-based API which is the primary API and where
  new features are added. This book is exclusively be about the
  DataFrame/Dataset-based API also known as Spark ML.
  
  However, both APIs provide functionalities to build classification, regression,
  clustering models and recommender systems in a distributed manner.
  
  We'll give a more thorough presentation of the DataFrame-based API later in
  this chapter.
  
  === Cluster managers
  
  Because Apache Spark is a distributed data processing framework, it relies on a
  cluster manager to scale to thousands of nodes. A cluster manager will try to
  satisfy Spark's requests for computing power based on which machines are
  available in the cluster.
  
  Three cluster managers are supported by Spark: Hadoop YARN, Apache Mesos and
  Standalone (which comes bundled with Spark).
  
  === Benefits of having those components being part of Spark
  
  There are two main advantages to having these subprojects as part of Spark:
  
  - You don't have to deal with version compatibility.
  - New features and performance improvements propagate almost instantaneously to
  the other parts of Spark.
  - You can use multiple components simultaneously. For example, you can
  interrogate a machine learning model in a streaming application.
  
  == What is Spark ML?
  
  If we dig deeper into the example we used when introducing machine learning,
  suppose you're working for AirBnB and in charge of implementing the suggestions
  for the daily rate you see when you want to list your apartment. How would you
  go about it? One way to do it would be to build a regression model which, as
  we've seen earlier in the chapter, can be summed to the construction of a
  function with parameters corresponding to the features of the accommodation such
  as which outputs a price.
  
  The function we want to obtain uses all the apartment's features as parameters
  and outputs the price we're going to suggest:
  
  `f(surface area, nb bedrooms, etc) = suggested price`
  
  Of course, we're looking for the best function that outputs a suggested price
  reflecting the reality of the market in the best possible way so we don't upset
  our users by overpricing or underpricing: if it's overpiced the apartment will
  never be filled. If it's underpriced, the people listing it will feel ripped off
  or the people looking for accommodation will assume hidden defects.
  
  Keep in mind that we will build this function by looking through the history
  of listings. There are more than 2 million active listings at any time on
  AirBnB, so we need to consider the pricing evolution the listings went through
  and all listings that are not active at that point in time. Moreover, we could
  imagine that AirBnB compiles a thorough list of an apartment's characteristics
  (not just the one we displayed earlier). You can guess that this amounts to a
  huge quantity of data. As a result, you won't be able to resort to a
  non-distributed machine learning librar because the data will be too large for
  your machine's RAM.
  
  Fortunately, Spark ML solves this issue because it's distributed. It will be
  able to process all our data and there are quite a few regression algorithms
  available out of the box. It is also relatively ease to use:
  the distributed nature of the algorithms is hidden from us. This implies that
  there is no cognitive overhead when leveraging a distributed algorithm compared
  to a non-distributed one.
  
  Spark ML supports many algorithms that touch quite a few machine learning
  domains such as classification, clustering or recommender systems not just
  regression.
  
  The main programming abstraction in Spark ML is the _Pipeline_, an idea borrowed
  from scikit-learn, a popular non-distributed Python machine learning library,
  