论文原文：A Berkeley View of Systems Challenges for AI

这又是篇不涉及过多技术的一篇论文，不过其中提到的Challenges确是目前AI Systems一直在努力解决的。

文中提到的几点关于AI System的Challenge：

• Design AI systems that learn continually by interacting with a dynamic environment, while making decisions that are timely, robust, and secure.
• Design AI systems that enable personalized applica- tions and services yet do not compromise users’ privacy and security.
• Design AI systems that can train on datasets owned by different organizations without compromising their confidentiality, and in the process provide AI capabilities that span the boundaries of potentially competing organization.
• Develop domain-specific architectures and soſtware systems to address the performance needs offuture AI applications in the post-Moore’s Law era, including custom chips for AI work-loads, edge-cloud systems to efficiently process data at the edge, and techniques for abstracting and sampling data.

在论文中也提出了一些解决上述Challenge的研究方向:

论文原文：MLSys: The New Frontier of Machine Learning Systems

这是一篇发表意义大于内容本身的一篇论文，它是一个全新的会议MLSys的开篇。所谓MLSys，顾名思义就是Machine Learning和System的交叉领域。

在论文中，作者从两个维度阐述了MLSys涉及的核心问题：
维度一：

• How should software systems be designed to support the full machine learning lifecycle, from program- ming interfaces and data preprocessing to output interpretation, debugging and monitoring?
• How should hardware systems be designed for machine learning?
• How should machine learning systems be designed to satisfy metrics beyond predictive accuracy, such as
  power and memory efficiency, accessibility, cost, latency, privacy, security, fairness, and interpretability?

维度二：

• high-level systems for ML that support interfaces and
  workflows for ML development—the analogue of traditional work on programming languages and software
  engineering.
• low-level systems for ML that involve hardware or software—and that often blur the lines
  between the two—to support training and execution of models, the analogue of traditional work on compilers and
  architecture.

这篇论文最流弊的地方就是它的作者署名(亮点自寻)：

术语解释

• AD: Active Directory。
• Service Session Key: 服务会话密钥。
• Logon Session Key: 登录会话密钥。
• KDC: Key Distribution Center。
• KAS: Key Kerberos Authentication Service。它是KDC的一个服务。
• TGS: Ticket Granting Service；它是KDC的一个服务。
• Service Ticket: 服务票据，通过TGS获取，主要包括用户信息与Service Session Key。
• TGT: Ticket Granting Ticket。通过KAS获取，主要包括用户信息与Logon Session Key。
• Authenticator: 交互双方预先知晓的信息，通过它来对对方做认证。

KDC整体结构图

Kerberos认证的流程

1. 客户端通过KDC的KAS服务获取TGT。
2. 通过TGT获取Service Ticket。
3. 通过Service Ticket访问服务资源 。

Client与Server之间的消息交互

1. 客户端发送消息到KDC的KAS服务一获取TGT
   
2. KAS 发送信息到Client
   
3. 客户端发送消息到KDC（Key Distribution Center）的TGS（Ticket Granting Service）服务一获取ST（Service Ticket）
   
4. TGS服务返回消息给客户端
   
5. 客户端通过ST访问服务器资源
   
6. 客户端对服务器的认证
   

之前BASE理论中提到，分布式系统对一致性做出了权衡。因此涌现出很多分布式协议。其中比较著名的是：两段式提交（Two Phase Commit）和三段式提交（Three Phase Commit）。

两段式提交

定义

每个参与者将操作成败通知协调者，再由协调者根据所有参与者的反馈情报，决定各参与者是否要提交操作还是中止操作。

所谓的两个阶段分别是：

第一阶段：准备阶段/投票阶段

1. 协调者节点向所有参与者节点询问是否可以执行提交操作，并开始等待各参与者节点的响应。
2. 参与者节点执行询问发起为止的所有事务操作，并将Undo信息和Redo信息写入日志。
3. 各参与者节点响应协调者节点发起的询问。如果参与者节点的事务操作实际执行成功，则它返回一个”同意”消息；如果参与者节点的事务操作实际执行失败，则它返回一个”中止”消息。

第二阶段：提交阶段/执行阶段

当协调者节点从所有参与者节点获得的相应消息都为”同意”时：

1. 协调者节点向所有参与者节点发出”正式提交”的请求。
2. 参与者节点正式完成操作，并释放在整个事务期间内占用的资源。
3. 参与者节点向协调者节点发送”完成”消息。
4. 协调者节点收到所有参与者节点反馈的”完成”消息后，完成事务。

如果任一参与者节点在第一阶段返回的响应消息为”终止”，或者 协调者节点在第一阶段的询问超时之前无法获取所有参与者节点的响应消息时：

1. 协调者节点向所有参与者节点发出”回滚操作”的请求。
2. 参与者节点利用之前写入的Undo信息执行回滚，并释放在整个事务期间内占用的资源。
3. 参与者节点向协调者节点发送”回滚完成”消息。
4. 协调者节点收到所有参与者节点反馈的”回滚完成”消息后，取消事务。

优缺点

• 优点：原理简单，实现方便。
• 缺点：同步阻塞，单点问题，数据不一致，容错性不好

三段式提交

定义

由于两段式提交存在诸多缺陷，因此研究者在两段式基础上进行改进，实现了三段式提交。与两阶段提交不同的是，三阶段提交有两个改动点：

• 引入超时机制 - 同时在协调者和参与者中都引入超时机制。
• 在第一阶段和第二阶段中插入一个准备阶段，保证了在最后提交阶段之前各参与节点的状态是一致的。

所谓的三个阶段分别是：

第一阶段：CanCommit

1. 协调者向参与者发送CanCommit请求。询问是否可以执行事务提交操作。然后开始等待参与者的响应。
2. 参与者接到CanCommit请求之后，正常情况下，如果其自身认为可以顺利执行事务，则返回Yes响应，并进入预备状态；否则反馈No。

第二阶段：PreCommit

1. 协调者向所有参与者节点发出 preCommit 的请求，并进入 prepared 状态。
2. 参与者受到 preCommit 请求后，会执行事务操作，对应 2PC 准备阶段中的 “执行事务”，也会 Undo 和 Redo 信息记录到事务日志中。
3. 如果参与者成功执行了事务，就反馈 ACK 响应，同时等待指令：提交（commit） 或终止（abort）。

第三阶段：Do Commit

1. 协调者接收到各参与者发送的ACK响应，那么他将从预提交状态进入到提交状态。并向所有参与者发送 doCommit 请求。
2. 参与者接收到 doCommit 请求之后，执行正式的事务提交。并在完成事务提交之后释放所有事务资源。
3. 事务提交完之后，向协调者发送 ACK 响应。
4. 协调者接收到所有参与者的 ACK 响应之后，完成事务。

优缺点

• 优点：相对于二阶段提交，三阶段提交主要解决的单点故障问题，并减少了阻塞的时间。
• 缺点：三阶段提交也会导致数据一致性问题。由于网络原因，协调者发送的 abort 响应没有及时被参与者接收到，那么参与者在等待超时之后执行了 commit 操作。这样就和其他接到 abort 命令并执行回滚的参与者之间存在数据不一致的情况。

参考资料

• https://en.wikipedia.org/wiki/Two-phase_commit_protocol
• https://en.wikipedia.org/wiki/Three-phase_commit_protocol
• https://juejin.im/post/5b26648e5188257494641b9f

BASE定理简介

BASE理论是Basically Available(基本可用)，Soft State（软状态）和Eventually Consistent（最终一致性）三个短语的缩写，由eBay架构师Dan Pritchett提出来的。

BASE是对CAP中一致性和可用性权衡的结果，其来源于对大规模互联网分布式系统实践的总结，是基于CAP定律逐步演化而来。其核心思想是即使无法做到强一致性，但每个应用都可以根据自身业务特点，才用适当的方式来使系统打到最终一致。

BASE理论的内容

基本可用（Basically Available）

所谓基本可用就是在出现不可预知的故障时，系统主体功能依然可用。一个比较典型的例子就是在电商促销时，为保护购物系统，部分消费有可能会被导到一个降级页面。

软状态（Soft State）

所谓软状态是指允许系统中的数据存在中间状态，并认为这种状态不影响系统的整体可用性。典型的例子如在分布式文件系统中，数据的写入往往是先写入一份，再异步生成多个副本（同步生成副本不属于这种情况）。

最终一致性（Eventually Consistent）

上面提到了软状态，但是系统不可以一直处于软状态，必须有一个期限。在期限过后，应当保证所有副本数据是的一致的，从而达到数据的最终一致性。而在实际工程实践中，最终一致性分为5种：

因果一致性（Causal consistency）

如果节点A在更新完某个数据后通知了节点B，那么节点B之后对该数据的访问和修改都是基于A更新后的值。于此同时，和节点A无因果关系的节点C的数据访问则没有这样的限制。

读己之所写（Read your writes）

节点A更新一个数据后，它自身总是能访问到自身更新过的最新值，而不会看到旧值。其实也算一种因果一致性。

会话一致性（Session consistency）

会话一致性将对系统数据的访问过程框定在了一个会话当中：系统能保证在同一个有效的会话中实现 “读己之所写” 的一致性，也就是说，执行更新操作之后，客户端能够在同一个会话中始终读取到该数据项的最新值。

单调读一致性（Monotonic read consistency）

单调读一致性指的是：如果一个节点从系统中读取出一个数据项的某个值后，那么系统对于该节点后续的任何数据访问都不应该返回更旧的值。

单调写一致性（Monotonic write consistency）

单调写一致性指的是：一个系统要能够保证来自同一个节点的写操作被顺序的执行。

在实际的实践中，这5种系统往往会结合使用，以构建一个具有最终一致性的分布式系统。事实上，最终一致性并不是只有那些大型分布式系统才涉及的特性，许多现代的关系型数据库都采用了最终一致性模型。在现代关系型数据库中，大多都会采用同步和异步方式来实现主备数据复制技术。在同步方式中，数据的复制过程通常是更新事务的一部分，因此在事务完成后，主备数据库的数据就会达到一致。而在异步方式中，备库的更新往往会存在延时，这取决于事务日志在主备数据库之间传输的时间长短，如果传输时间过长或者甚至在日志传输过程中出现异常导致无法及时将事务应用到备库上，那么很显然，从备库中读取的数据将是旧的，因此就出现了数据不一致的情况。当然，无论是采用多次重试还是人为数据订正，关系型数据库还是能够保证最终数据达到一致。

总结

总的来说，BASE理论面向的是大型高可用可扩展的分布式系统，和传统事务的ACID特性是相反的，它完全不同于ACID的强一致性模型，而是提出通过牺牲强一致性来获得可用性，并允许数据在一段时间内是不一致的，但最终达到一致状态。

参考资料

• https://juejin.im/post/5b2663fcf265da59a401e6f8
• https://xinklabi.iteye.com/blog/2341034

综述

这篇文章记录了Spark应用开发过程遇到的各种问题及解决方案。主要来自于个人开发的实践，也有部分解决方案来自互联网，如有不足之处欢迎批评指正。本人会持续更新转载请保留原文地址

问题描述及解决方案

java.lang.OutOfMemoryError: Java heap space错误

这个是Spark应用常见错误。JVM堆内存空间不足。解决方案如下：

• 首先要判断是Driver或者Executor出现OOM，通过–driver-memory或者–executor-memory进行调整。
• 如果是Spark SQL或者Spark Streaming的程序，建议适当地提高heap size。

java.lang.OutOfMemoryError: GC overhead limit exceeded错误

这个也是Spark应用常见错误，由于GC时间过长导致的。解决方案如下：

• 直接通过–driver-memory或者–executor-memory增加heap size
• 修改GC policy。可以使用-XX:UseG1GC或者-XX:UseParallelGC

编译OK，运行时出NoClassDefineError错误

这个错误非常清晰，根本原因就是jar没有放入classpath之中。首先需要判断到底是Driver还是Executor缺少这个jar包。

• 将jar包路径配置到spark.driver.extraClassPath或者spark.executor.extraClassPath。
• 将jar包路径通过spark-submit的–driver-class-path或者–executor-class-path指定。

其实，spark-submit还有一个–packages参数，这个参数让Spark通过Maven从本地或者远程的repository处获取jar包。这个参数看似非常方便，但实际使用的时候不是很实用。因为Hadoop和Spark集群应用一般都是部署在内网的，为了数据安全，一般情况都是无法访问外网的。

org.apache.spark.SparkException: Task not serializable

关于这个问题有单独的一篇文章进行分析，详见：Spark Troubleshooting - Task not serializable问题分析

java.io.IOException: No space left on device错误

具体stack trace如下：

1
2
3
4
5
6
7

stage 89.3 failed 4 times, most recent failure:
Lost task 38.4 in stage 89.3 (TID 30100, rhel4.cisco.com): java.io.IOException: No space left on device
            at java.io.FileOutputStream.writeBytes(Native Method)
            at java.io.FileOutputStream.write(FileOutputStream.java:326)
            at org.apache.spark.storage.TimeTrackingOutputStream.write(TimeTrackingOutputStream.java:58)
            at java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:82)
            at java.io.BufferedOutputStream.write(BufferedOutputStream.java:126)

1
2
3

Directory to use for "scratch" space in Spark, including map output files and RDDs that get stored on disk.
This should be on a fast, local disk in your system.
It can also be a comma-separated list of multiple directories on different disks.

spark.driver.maxResultSize超出错误

数据拉回Driver端是有限制的，通过spark.driver.maxResultSize控制：

• 默认是1g
• 可以设置为0或者unlimited
• 如果设置成unlimited就不会再遇到这个错误，取而代之的是OOM。

java.lang.IllegalArgumentException: Size exceeds Integer.MAX_VALUE

具体stack trace如下：

1
2
3
4
5
6
7
8
9

java.lang.IllegalArgumentException: Size exceeds Integer.MAX_VALUE
at sun.nio.ch.FileChannelImpl.map(FileChannelImpl.java:828) at
org.apache.spark.storage.DiskStore.getBytes(DiskStore.scala:123) at
org.apache.spark.storage.DiskStore.getBytes(DiskStore.scala:132) at
org.apache.spark.storage.BlockManager.doGetLocal(BlockManager.scala:51 7) at
org.apache.spark.storage.BlockManager.getLocal(BlockManager.scala:432) at
org.apache.spark.storage.BlockManager.get(BlockManager.scala:618) at
org.apache.spark.CacheManager.putInBlockManager(CacheManager.scala:146 ) at
org.apache.spark.CacheManager.getOrCompute(CacheManager.scala:70)

1

val buf = ByteBuffer.allocate(length.toInt) * ByteBufferislimitedbyInteger.MAX_SIZE

参考资料

• http://spark.apache.org/docs/latest/configuration.html

CAP定理简介

在理论计算机科学中，CAP定理（CAP theorem），又被称作布鲁尔定理（Brewer’s theorem），指的是在一个分布式系统中，Consistency（一致性）、 Availability（可用性）、Partition tolerance（分区容错性）这三个基本需求，最多只能同时满足其中的2个

• 一致性（Consistency）：数据在多个副本之间能够保持一致的特性
• 可用性（Availability）：系统提供的服务必须一直处于可用的状态，每次请求都能获取到非错的响应（不保证获取的数据为最新数据）
• 分区容错性（Partition tolerance）：分布式系统在遇到任何网络分区故障的时候，仍然能够对外提供满足一致性和可用性的服务，除非整个网络环境都发生了故障

CAP权衡

既然根据CAP定理，我们无法同时满足一致性，可用性和分区容错性，那要舍弃哪个呢？

• CA without P
  如果不要求P（不允许分区），则C（强一致性）和A（可用性）是可以保证的。但其实分区不是你想不想的问题，而是始终会存在，因此CA的系统更多的是允许分区后各子系统依然保持CA。
• CP without A
  如果不要求A（可用性），相当于每个请求都需要在Server之间强一致，而P（分区）会导致同步时间无限延长，如此CP也是可以保证的。很多传统的数据库分布式事务都属于这种模式。
• AP wihtout C
  要高可用并允许分区，则需放弃一致性。一旦分区发生，节点之间可能会失去联系，为了高可用，每个节点只能用本地数据提供服务，而这样会导致全局数据的不一致性。实际上，目前大部分NoSQL都属于这一类。

参考文档

• https://en.wikipedia.org/wiki/CAP_theorem

问题描述

出现“org.apache.spark.SparkException: Task not serializable”这个错误，一般是因为在map、filter等的参数使用了外部的变量，但是这个变量不能序列化。其中最普遍的情形是：当引用了某个类（经常是当前类）的成员函数或变量时，会导致这个类的所有成员（整个类）都需要支持序列化。虽然许多情形下，当前类使用了“extends Serializable”声明支持序列化，但是由于某些字段不支持序列化，仍然会导致整个类序列化时出现问题，最终导致出现Task未序列化问题。

解决办法与编程建议

这个问题主要是引用了某类的成员变量或函数，并且相应的类没有做好序列化处理导致的。因此解决这个问题无非以下两种方法：

不在（或不直接在）map等闭包内部直接引用某类成员函数或成员变量

• 对于依赖某类成员变量的情形
  如果程序依赖的值相对固定，可取固定的值，或定义在map、filter等操作内部，或定义在scala object对象中。
  如果依赖值需要程序调用时动态指定（以函数参数形式），则在map、filter等操作时，可不直接引用该成员变量，而是根据成员变量的值重新定义一个局部变量，这样map等算子就无需引用类的成员变量。

• 对于依赖某类成员函数的情形
  如果函数功能独立，可定义在scala object对象中（类似于Java中的static方法），这样就无需一来特定的类。

如果引用了某类的成员函数或变量，则需对相应的类做好序列化处理

首先该类继承Serializable接口，然后对于不能序列化的成员变量使用“@transent”标注，告诉编译器不需要序列化。 此外如果可以，可将依赖的变量独立放到一个小的class中，让这个class支持序列化，这样做可以减少网络传输量，提高效率。

参考资料

• https://blog.csdn.net/javastart/article/details/51206715

概述

在我们编写Spark Application的的过程中，会涉及到很多概念，下面会介绍一些基本概念用于更好地理解Spark的设计以及应用开发。

RDD（Resillent Distributed Dataset）

弹性数据集。它是Spark的基本数据结构，Spark中的所有数据都是通过RDD的形式进行组织。RDD是不可变的数据集合，不可变的意思是RDD中的每个分区数据是只读的。RDD数据集是要做逻辑分区的，每个分区可以单独在集群节点进行计算。

DAG（Directed Acycle Graph）

有向无环图。在图论中，如果一个有向图无法从有一个定点出发经过若干条边回到该点，则这个图是有向无环图。Spark中用DAG来表示RDD之间的血缘关系。

NarrowDependency

窄依赖，子RDD依赖于父RDD中固定的Partition。NarrowDependency分为OneToOneDependency和RangeDependency。

ShuffleDependency

Shuffle依赖，也称作宽依赖，子RDD可能对父RDD中所有的Partition都有依赖。子RDD对父RDD各个Partition的依赖取决于分区计算器（Partitioner）的算法。

Job

所谓一个Job，就是由一个RDD的action触发的动作，可以简单的理解为，当你需要执行一个RDD的action的时候，会生成一个job。

Stage

Stage是一个Job的组成单位，就是说，一个Job会被切分成1个或多个的Stage，然后各个Stage会按照执行顺序依次执行。至于Job根据什么标准来切分Stage，可以简单理解为按照是否需要shuffle来划分stage。当一个操作需要shuffle时，Spark就会将其划分为一个Stage。具体细节后续会有文章单独介绍。

Task

Task是Spark的执行单元，每个Stage都会包含一个或者一组Task。一般来说，数据有多少个partition就会有多少个Task。

Partition

Partition类似Hadoop的Split，是数据的一种划分方式，计算以Partition为单位进行的。Partition的划分依据有很多，也可以自己定义。例如HDFS的文件的划分方式就是按照HDFS Block大小来划分的。

Shuffle

Shuffle是Spark Application中一个非常重要的阶段。Shuffle用于打通map task（在Spark称为ShuffleMapTask）的输出与reduce task的任务（在Spark中就是ResultTask）的输入，map task的输出结果按照指定的分区策略分配给某一个分区的reduce task。

Spark 架构简介

Spark是一个master/slave架构的分布式系统，它的架构主要包含有

• Spark Driver
• Spark Executor
• Cluster Manager
  

一个Spark集群一般拥有单个的Driver和多个的Executor。Spark Driver和Executor都是独立运行的JVM进程，它们可以运行在单台机器上，也可以运行在多台机器上。

Spark Driver

Spark Driver是一个Spark Application的主入口，它可以用Scala，Python或者R进行编写。一个Spark Driver包含有一个SparkContext，这是整个Spark Application中最核心的组件。同时，还包含有DAGScheduler, TaskScheduler, BackendScheduler和BlockManager等组件用于将用户代码转换为Spark job运行在集群当中。Spark Driver的主要功能包括：

• 负责协调Job的运行和以及Cluster Manager进行交互。
• 将RDD转换为执行的DAG图，同时把DAG图分为不同的Stage
• 将Job切割成更小的执行单元，Task，由Executor执行。
• 启动一个HTTP Server，端口为4040。这个Web UI会把Spark Application运行时的信息展示出来。

Spark Executor

Spark Executor是Task的实际执行者。每个Application的Executor数量可以通过配置指定（Static Allocation）或者有Spark动态分配（Dynamic Allocation）。Executor的主要功能包括：

• 负责所有的数据处理工作
• 用于读取和写入外部数据源
• 缓存着计算过程中的数据

Cluster Manager

严格上说Cluster Manager并不是Spark的一部分，而是一个外部的Service（除了Standalone）。Spark Driver会和其进行交互用于从集群里获取资源（CPU，Memory等）。目前Spark支持4种Cluster Manager：

• Standalone：这是一种Spark自带的集群管理模式，设计也比较简单。
• Apache Mesos：Mesos是一种通用的集群资源管理服务，用于管理MapReduce应用或者其他类型的应用。
• Hadoop YARN ：YARN是由Hadoop 2.0引入的集群资源管理服务。
• Kubernetes：Kubernetes是一种管理containerized的应用的服务。Spark 2.3以后引入了对Kubernetes的支持。

至于选择使用哪一种Cluster Manager，完全取决于生产环境以及业务场景，并没有绝对的优劣。