10 Essential Spark Interview Questions *

The main method, calculate, reads two sets of data. (在本例中，它们是从一个常量内联数据结构中提供的，该数据结构使用 parallelize.)应用于它们的映射将它们转换成元组，每个元组由a组成 userId and the object itself. The userId is used to join the two datasets.

连接的数据集由所有登录活动的用户过滤. It is then transformed into a tuple consisting of userName and the event timestamp.

这最终减少到每个用户的第一个登录条目，并写入控制台.

The result will be:

Doe, John: 1514764800
Doe, Jane: 1514808000

The complete code should look like this:

def calculate(sparkSession: SparkSession): Unit = {
 val UserIdColName = "userId"
 val UserNameColName = "userName"
 val CountColName = "totalEventCount"

 val userRdd: DataFrame = readUserData(sparkSession)
 userActivityRdd: DataFrame = readUserActivityData(sparkSession)


 val result = userRdd
   .repartition(col(UserIdColName))
   .join(userActivityRdd, UserIdColName)
   .select(col(UserNameColName))
   .groupBy(UserNameColName)
   .count()
   .withColumnRenamed("count", CountColName)

 result.show()
}

The join expression can take different kinds of parameters. The following alternatives produce the same result:

.join(userActivityRdd,UserIdColName, "inner")

.join(userActivityRdd, Seq(UserIdColName))

.join(userActivityRdd, Seq(UserIdColName), "inner")

将“left”作为最后一个参数将返回错误的结果(没有事件的用户将被包括在内), showing an event count of 1).

传递“right”作为最后一个参数将返回正确的结果，但会在语义上产生误导.

   .groupBy(UserNameColName)

This is required. Without it, the total number of rows would be counted and result would be a Long instead of a DataFrame (so the code wouldn’t even compile, since the show() method does not exist for Long.)

   .count()

This is the actual aggregation that adds a new column (count) to the DataFrame.

   .withColumnRenamed("count", CountColName)

This renames the count column to totalEventCount, as requested in the question.

注册UDF的那行应该替换为下面的代码片段:

def calculate(sparkSession: SparkSession): Unit = {
...


 val categoryNodesWithChildrenBC = sparkSession.sparkContext.broadcast(categoryNodesWithChildren)

 sparkSession.udf.register("isChildOf", (nodeId: Int, parentNodeId: Int) =>
   nodeId != parentNodeId && categoryNodesWithChildrenBC.value.getOrElse(nodeId, Set[Int]()).contains(parentNodeId))

...

}

第一种方法的问题是，它使用了驱动程序应用程序中不可用的变量 per se on the worker nodes. Spark will fetch the variable (meaning, the whole Map) from the master node each time the UDF is called. 这可能导致主服务器的负载非常高，整个集群可能变得无响应. 对代码的修改会将变量的副本发送(广播)到每个工作节点，在这些节点中，变量可以作为对象进行访问 org.apache.spark.broadcast object that holds the actual Map.

The basic structure of a Spark-cluster:

集群管理器不是Spark框架本身的一部分——尽管Spark自带了它自己的框架, this one should not be used in production. Supported cluster managers are Mesos, Yarn, and Kybernetes.

驱动程序是在Spark Master上执行的Java、Scala或Python应用程序.

As part of the program, some Spark framework methods will be called, which themselves are executed on the worker nodes.

每个工作节点可能运行多个执行器(按配置:通常每个可用CPU内核一个执行器). 每个执行程序都将从要执行的调度器接收一个任务.

The modules of Apache Spark run directly on top of its core:

Apache Spark (Core)

Spark由一个核心框架组成，该框架管理数据的内部表示，包括:

• serialization
• memory allocation
• caching
• increasing resilience by storing intermediate snapshots on disk
• automatic retries
• data exchange (shuffling) between worker nodes
• etc.

It also provides a bunch of methods to transform data (like map and reduce). 所有这些方法都适用于弹性分布式数据集(RDD).

Spark自动识别单个步骤之间的依赖关系，从而知道哪些步骤可以并行执行.

This is accomplished by building a directed acyclic graph (DAG), 这也意味着转换不会立即执行, but when action functions are called.

所以基本上，方法可以分为两种类型:RDD转换和操作.

These are RDD transformations:

• map(func)
• flatMap()
• filter(func)
• mapPartitions(func)
• mapPartitionWithIndex()
• union(dataset)
• intersection(dataset)
• distinct()
• groupByKey()
• reduceByKey(func, [numTasks])
• sortByKey()
• join()
• coalesce()

And these are RDD actions:

• count()
• collect()
• take(n)
• top()
• countByValue()
• reduce()
• fold()
• aggregate()
• foreach()

在Spark UI web界面中可以查看正在运行的作业的DAG. 它还显示挂起的作业、任务列表以及当前的资源使用和配置.

如果配置了历史服务器，还可以查看已完成(或失败)作业的大多数信息.

Spark SQL

This is an abstraction of Spark’s core API. Whereas the core API works with RDD, and all transformations are defined by the developer explicitly, Spark SQL represents the RDD as so-called DataFrames. The DataFrame API is more like a DSL that looks like SQL.

开发人员甚至可以通过将DataFrame注册为已命名的内存表来更加抽象RDD. 然后可以像使用SQL查询关系数据库中的表一样查询该表.

Spark Streaming

它可以轮询分布式日志，如Apache Kafka或Amazon Kinesis(以及其他一些消息传递系统), like ActiveMQ) to process the messages in micro-batches. (几乎)所有Spark批处理作业可用的功能也可以应用到Spark Streaming提供的RDD上.

MLlib

MLlib提供了在一般数据分析(如聚类和回归)和机器学习中常用的高级算法. It provides the functionality to define pipelines, train models and persist them, and read trained models to apply them to live data.

GraphX

这允许您将RDD表示为图(节点通过边连接)并对其执行一些基本的图操作. Currently (only) three more advanced algorithms are provided: PageRank, ConnectedComponents, and TriangleCounting.

The missing SQL should look something like this:

 val result = sparkSession
   .sql(s"SELECT u.userName, MIN(ua.timestampEpochMs) AS firstLogin " +
     s"FROM $UserTableName u " +
     s"JOIN $UserActivityTableName ua ON u.userId=ua.userId " +
     s"WHERE ua.activityTypeId=$LoginActivityTypeId " +
     s"GROUP BY u.userName")

(The $ notation是Scala的一个特性，用于替换字符串中的表达式，而表名也可能是“硬编码”的. 请注意，永远不要对用户提供的变量这样做，因为这会使代码容易受到注入漏洞的攻击.)

The call of this function is performed by the driver application. The assignment to the result 值是DAG的定义，包括它的执行，由 collect() call. All parts of this (including the logic of the function mapDateTime2Date) are executed on the worker nodes.

The resulting value that is stored in result 是在主服务器上收集的数组，因此对该值执行的映射在主服务器上运行.

下表概述了每个框架的一些特征. 何时使用哪个框架并不总是有一个明确的答案. 特别是因为他们经常在不重要的细节上有所不同.

But it’s important to understand:

• Kafka Streams只是一个库(没有额外的基础设施组件), 但是它有责任部署和扩展流应用程序).
• 在低延迟流处理方面，Flink是目前最优越/功能最丰富的框架(当流被用作服务之间的实时核心通信时，这一点很重要)。.
• Spark的主要优点是整个现有的生态系统，包括MLlib/GraphX抽象，部分代码可以被批处理和流处理功能重用.

def calculate(sparkSession: SparkSession): Unit = {

 val pageRdd: RDD[(VertexId, Page)] =
   readPageData(sparkSession)
     .map(e => (e.pageId, e))
     .cache()
 val pageReferenceRdd: RDD[Edge[PageReference]] = readPageReferenceData(sparkSession)

 val graph = Graph(pageRdd, pageReferenceRdd)
 val PageRankTollerance = 0.005
 val ranks = graph.pageRank(PageRankTollerance).vertices

 ranks.take(1000)
   .foreach(println)
}

A graph consists of Vertex objects and Edge objects that are passed to the Graph object as RDDs of type RDD[VertexId, VT] and RDD[Edge[ET]] (where VT and ET 是否有任何用户定义的类型应该与给定相关联 Vertex or Edge). The constructor of the Edge type is Edge[ET](srcId: VertexId, dstId: VertexId, attr: ET). The type VertexId is basically an alias for Long.

The repartition 语句生成10个分区(无论从任何地方加载分区时是多还是少). 对于庞大的数据集，这些变量可能会变得非常大，并且可能无法容纳为一个执行器分配的内存.

Also, only one partition can be allocated per executor. This means, 240个执行器中只有10个被使用(10个节点和24个核心), each running one executor).

If the number is chosen too high, 调度器管理分区的开销会增加并降低性能. 在某些情况下，对于非常小的分区，它甚至可能超过执行时间本身.

The recommended number of partitions is between two to three times the number of executors. In our case, 600 = 10 x 24 x 2.5 would be an appropriate number of partitions.

Spark MLlib has two basic components: Transformers and Estimators.

A Transformer reads a DataFrame and returns a new DataFrame with a specific transformation applied (e.g. new columns added). An Estimator is some machine learning algorithm that takes a DataFrame to train a model and returns the model as a Transformer.

Spark MLlib允许您将多个转换组合到一个管道中 apply complex data transformations:

The following image shows such pipeline for training a model:

The model produced can then be applied to live data: