CS100.1x Introduction to Big Data with Apache Spark

g87616758

CS100.1x简介
　　这门课主要讲数据科学，也就是data science以及怎么用Apache Spark去分析大数据。

Course Software Setup
　　这门课主要介绍如何编写和调试PySpark。本节主要介绍环境搭配。为了让所有人环境一致，本课程的编程环境是用Virtual Machine。你需要安装VirtualBox和Vagrant来搭环境。

硬件和软件要求
　　这门课需要的最小硬件配置如下：

• 硬盘空间： 3.5 GB　　
  
  
  
• 内存： 2.5 GB (4+ GB 更好)　　
  
  
  
• 处理器： 任何Intel和AMD的多核处理器都行
  

　　所支持的系统：

• 64-bit (preferred) Windows 7 or later
  
• 64-bit (preferred) Mac OS X 10.9.5 or later
  
• 64-bit (preferred) Linux (CentOS 6 or later, or Ubuntu 14.04 or later)
  
• 32-bit Windows 7 or later
  
• 32-bit Linux (CentOS 6 or later, or Ubuntu 14.04 or later)
  

　　对Linux用户来说，Vagrant最好用1.7以上的版本，默认的Vagrant可能不支持课程环境。
　　补充一下：视频里老师用的Windows8.

安装软件
　　我们需要安装两个免费的软件VirtualBox和Vagrant。确保VirtualBox版本为4.3.28以及以上，Vagrant 版本为1.7.2以及以上。
　　我在这里装了好几遍，最后成功的版本是VirtualBox-4.3.40-110317-Win和vagrant_1.7.2。安装成功之后，执行如下操作，我在这里直接复制课件：

• Create a custom directory (e.g., for windows users c:\users\marco\myvagrant or for Mac/Linux users /home/marco/myvagrant)
  
• Download this file (https://github.com/spark-mooc/mooc-setup/archive/master.zip) to the custom directory and unzip it.
  
• From the unzipped file, copy Vagrantfile to the custom directory you created in step #1 (NOTE: It must be named exactly "Vagrantfile" with no extension)
  
• Open a DOS prompt (Windows) or Terminal (Mac/Linux), change to the custom directory, and issue the command "vagrant up --provider=virtualbox"
  

　　看各位的网速，我这里平均80KB/S下载了大概2小时就装好了。

使用VM指南

• 启动。在dos或者terminal下进入c:\users\marco\myvagrant（其他系统类似）下运行vagrant up。
  
• 停止。在上述目录下运行vagrant halt。
  
• 删除。这是把整个环境删除的命令，vagrant destroy
  
• 一旦VM开始启动，在浏览器输入"http://localhost:8001/"或者"http://127.0.0.1:8001/"就进入到Notebook界面。如下图所示　　
  
  
  

运行你的第一个Notebook
　　这里教你如何运行第一个Notebook。

• 启动VM
  
• 从前面下载的mooc-setup-master.zip文件里找到lab0_student.ipynb（我的github上也有）
  
• 浏览器进入"http://localhost:8001/"或者"http://127.0.0.1:8001/"
  
• 点击upload，把该文件传上去
  
• 运行每个cell
  

lecture1 Introduction to Big Data and Data Science
　　本节主要讲了课程目标，数据分析简史以及三个大数据的例子，最后重点说了一下大数据的数据来源。干货不多，简单略过。

lecture2 Performing Data Science and Preparing Data
　　本节主要讲什么是数据科学，和数据科学相关的几个概念，比如数据库、机器学习。然后介绍了一些大牛对如何开展数据科学的观点，以及云计算。最后介绍了数据科学的常见主题和数据科学所发挥的作用。
　　数据科学又是一个新的、说不完的话题，这里不再多说，附上数据科学技能图一张。
　　


Lecture3 Big Data, Hardware Trends, and Apache Spark

The big data problem
　　大数据时代到来，之前用的处理数据的工具比如unix shell、R等只能在单机上跑，但是随着数据量越来越大，单机的计算和存储速度已经不能满足人们的需求，此时唯一的出路就是分布式计算。不过用廉价机器来组成集群的分布式，也存在这诸多问题，比如某台机器运行失败，网络传输慢以及各个机器性能不均匀等。这就需要我们从软件的层面来解决问题

Distributing Work
　　这里讲了一个分布式非常经典的例子word count。
　　对与在单机上统计的word count程序，我们只需要遍历一遍，把结果存储在一个哈希表里就行。
　　而假如这个文件非常大时，我们可能需要利用分布式计算框架来计算了。当文件很大时，我们把文件分成很多小块到不同的机器，然后分别对每个小块进行统计和计算。当所有的计算结果结束时，我们把这些结果给到一台机器去合并。这里又有个问题就是，假如这个结果也非常大怎么办，那我们就把相同的单词合并到一个机器（其实就是reduced过程）。
　　word count是一般是入门分布式的第一个例子，类似于学一门语言的hello world程序。我这里讲的很简单，有兴趣的同学可以搜搜看，例子满大街都是。
　　有了上面的结果，我们就能做出类似排序的操作了，我们可以知道哪个单词出现的频率最多等等。这里有存在两个问题。

• 我们如何把数据分发到不同的机器
  
• 如何处理某块数据处理失败的问题
  

　　先说后面一个问题。当数据处理失败时，最简单直接的方法就是重新开启一个任务处理这块数据。那假如数据处理慢怎么办，那就直接把这个任务kill掉，换太机器处理这块数据，因为这台机器可能不行了。

MapReduce and Spark
　　在MapReduce的过程中，每一个步骤都涉及到大量的I/O操作。map步骤要先从硬盘上读取数据，处理后，然后把结果存储在硬盘上；reduce操作就要把这些结果从硬盘读出来，处理后又写入硬盘。而这个MapReduce过程会一直重复这几个步骤，这就在I/O上消耗大量的操作，更坑的是I/O处理过程非常慢，所以MapReduce就因为I/O而显得很慢，特别是处理迭代多次的任务。由此诞生了Spark。
　　随着内存的价格下降，这意味着，我们可以把更多的数据放到内存中处理，而不是硬盘上。Spark也就是基于这个思想。理论上来说，Spark要比MapReduce快10到100倍。
　　Spark会提供一种分布式的包装对象RDD，我们通过RDD来进行各种各样的操作。这里可以把RDD简单理解成一种分布式的数据集合。Spark把分布式背后的细节都隐藏了，我们不需要去处理失败的任务以及特别慢的任务。
　　Spark无论从功能还是速度上都要优于MapReduce，所以欢迎大家弃MapReduce坑，来Spark这个坑。

Lecture4 Spark Essentials

Python Spark和RDD
　　这节我们将来学习Python Spark编程，推荐大家看看其API,非常全，例子也非常多。
　　Spark提供的Python编程接口也叫PySpark。一个Spark程序包括两个程序：driver program和workers program。前者运行在driver machine，后者运行在cluster，RDD则分布在workers上。
　　Spark程序的第一步是创建一个SparkContext，PySpark shell会自动帮你创建好sc变量。
　　在Spark中，RDD是不可变的，这是Spark的内在机制决定的，主要是为了更有效的跟踪节点信息和确保操作并行处理。我们只能通过：1.并行化一个python集合、2.转换存在的RDD、3.从其他分布式存储系统上读取，这三种方式获得RDD。有关RDD操作，下一节讨论。

work with RDD

create an RDD
　　

data = [1,2,3,4,5]　　
rDD = sc.parallelize(data, 4)
　　

　　这段代码通过python的list创建了一个RDD，并且分成了4分。
　　

distFile = sc.textFile("README.MD",4)　　

　　这段代码通过读取HDFS上文件来创建RDD

Spark transformations
　　这里的例子均来自api里的例子。分别是map，filter，distinct和flapMap操作
　　

>>> rdd = sc.parallelize(["b", "a", "c"])　　
>>> sorted(rdd.map(lambda x: (x, 1)).collect())
　　
[('a', 1), ('b', 1), ('c', 1)]
　　

>>> rdd = sc.parallelize([1, 2, 3, 4, 5])　　
>>> rdd.filter(lambda x: x % 2 == 0).collect()
　　
[2, 4]
　　

>>> sorted(sc.parallelize([1, 1, 2, 3]).distinct().collect())　　
[1, 2, 3]
　　

>>> rdd = sc.parallelize([2, 3, 4])　　
>>> sorted(rdd.flatMap(lambda x: range(1, x)).collect())
　　
[1, 1, 1, 2, 2, 3]
　　
>>> sorted(rdd.flatMap(lambda x: [(x, x), (x, x)]).collect())
　　
[(2, 2), (2, 2), (3, 3), (3, 3), (4, 4), (4, 4)]
　　

Spark action
　　由于Spark是在transformations采用lazy策略，所以进行transformations时，并不会真正的计算，只有碰到action操作时，才会计算。常见的action操作有reduce，take，collect和takeOrdered，下面是例子。collect()没有给例子，因为它就是直接返回所有的值。
　　

>>> from operator import add　　
>>> sc.parallelize([1, 2, 3, 4, 5]).reduce(add)
　　
15
　　
>>> sc.parallelize((2 for _ in range(10))).map(lambda x: 1).cache().reduce(add)
　　
10
　　
>>> sc.parallelize([]).reduce(add)
　　
Traceback (most recent call last):
　　...
　　
ValueError: Can not reduce() empty RDD
　　

>>> sc.parallelize([2, 3, 4, 5, 6]).cache().take(2)　　
[2, 3]
　　
>>> sc.parallelize([2, 3, 4, 5, 6]).take(10)
　　
[2, 3, 4, 5, 6]
　　
>>> sc.parallelize(range(100), 100).filter(lambda x: x > 90).take(3)
　　
[91, 92, 93]
　　

>>> sc.parallelize([2, 3, 4, 5, 6]).cache().take(2)　　
[2, 3]
　　
>>> sc.parallelize([2, 3, 4, 5, 6]).take(10)
　　
[2, 3, 4, 5, 6]
　　
>>> sc.parallelize(range(100), 100).filter(lambda x: x > 90).take(3)
　　
[91, 92, 93]
　　

>>> sc.parallelize([10, 1, 2, 9, 3, 4, 5, 6, 7]).takeOrdered(6)　　
[1, 2, 3, 4, 5, 6]
　　
>>> sc.parallelize([10, 1, 2, 9, 3, 4, 5, 6, 7], 2).takeOrdered(6, key=lambda x: -x)
　　
[10, 9, 7, 6, 5, 4]
　　

Caching RDDs
　　为了防止重复的读取数据，我们可以用cache()来缓存数据。上面的例子中有用到这个函数。

Spark key-value RDDs
　　Spark同样支持key-value。在PySpark中，用tuple来实现。相关函数有reduceByKey()，sortByKey()，groupByKey()
　　

>>> from operator import add　　
>>> rdd = sc.parallelize([("a", 1), ("b", 1), ("a", 1)])
　　
>>> sorted(rdd.reduceByKey(add).collect())
　　
[('a', 2), ('b', 1)]
　　

>>> tmp = [('a', 1), ('b', 2), ('1', 3), ('d', 4), ('2', 5)]　　
>>> sc.parallelize(tmp).sortByKey().first()
　　
('1', 3)
　　
>>> sc.parallelize(tmp).sortByKey(True, 1).collect()
　　
[('1', 3), ('2', 5), ('a', 1), ('b', 2), ('d', 4)]
　　
>>> sc.parallelize(tmp).sortByKey(True, 2).collect()
　　
[('1', 3), ('2', 5), ('a', 1), ('b', 2), ('d', 4)]
　　
>>> tmp2 = [('Mary', 1), ('had', 2), ('a', 3), ('little', 4), ('lamb', 5)]
　　
>>> tmp2.extend([('whose', 6), ('fleece', 7), ('was', 8), ('white', 9)])
　　
>>> sc.parallelize(tmp2).sortByKey(True, 3, keyfunc=lambda k: k.lower()).collect()
　　
[('a', 3), ('fleece', 7), ('had', 2), ('lamb', 5),...('white', 9), ('whose', 6)]
　　

>>> rdd = sc.parallelize([("a", 1), ("b", 1), ("a", 1)])　　
>>> sorted(rdd.groupByKey().mapValues(len).collect())
　　
[('a', 2), ('b', 1)]
　　
>>> sorted(rdd.groupByKey().mapValues(list).collect())
　　
[('a', [1, 1]), ('b', [1])]
　　

　　这里要补充的是，reduceByKey和aggregateByKey性能要比groupByKey高的多，所以尽量不用groupByKey。

Lecture 5 Semi-Structured Data

Key Data Management Concepts
　　这里先重点介绍了两个概念：data model和schema。前者的意思是描述数据的概念集合，后者的意思是对给定的数据模型，一种对特定数据的描述方法。（似乎翻译的不太好，囧，对数据库熟悉的应该很熟悉这两个概念）然后引出了the structure spectrum。
　　


Semi-Structured Tabular Data
　　这里介绍了file，table，cell。一般来说，这种数据存在这诸多的问题。课程讲了很多，总结来说，就是相比于关系型数据库，这种Semi-Structured Data显得杂乱无章（对比之下）。
　　在单机的情况下，我们一般用pandas的DataFrame来处理这些数据，R中也有类似的数据结构。
　　Spark在1.3后也引进了这种结构，作为RDD的延伸。PySpark的DataFrame和Spark的DataFrame可以相互转换。
　　RDD和DataFrame的性能差异非常大，在单机上，对比Spark DataFrame和PySpark的RDD，前者几乎要快5倍。所以在处理特别大的数据时，优先考虑用DataFrame。
　　后面重点分析了log files的格式和应用，都比较浅显，需要注意的是对文件的读写，Scala和Java的性能均要比Python好很多。

Lecture 6 Structured Data

Relational Database
　　所谓关系型数据库简单的说，就是建立在关系模型基础上的数据库。而上文提到的schema就是在一个关系型数据库里面，定义了表、每个表的字段，还有表和字段之间的关系。
　　关系型数据库的优缺点总结在下图。这是课程里截的图。这种对比在很多讲非关系型数据库的书或者博客里会提到。课程中还讲了SQL入门知识，这么不多讲了。
　　


joins in Spark
　　在Spark中，SparkSQL和Spark DataFrames的join()支持：inner，outer，left outer，right outer和semijoin。
　　对于 pair RDDs，PySpark支持：inner join, left outer join, right outer join, full outer join。其函数分别是join(), leftOuterJoin(), rightOuterJoin(),fullOuterJoin()。

Lecture7 Data Quality
　　这一节主要包括data cleaning, data quality, data gathering, data integration等内容。
　　有关这部分内容，我个人觉得和具体的业务场景有很大联系，不同的业务场景对数据清洗，数据质量检测，数据融合等都有不同的侧重点。
　　而至于一些常规的数据质量问题，一般有固定的清洗方法。但是数据质量问题在实际中是常态，一般的流程如下图所示。
　　

　　这中间涉及到数据传输，数据校验，统计学，数据库，元数据，数据编码，数据质量衡量等知识点，不再赘述。

Lecture8 Exploratory Data Analysis and Machine Learning
　　这一节终于涉及到一些数据分析和机器学习的东西了。内容也比前一节有干货。

Statistics, Business Questions, and Learning Techniques
　　我们在具体实际中，经常碰到一些统计问题，比如什么产品卖的最好，产品和时间有什么相关关系，亦或是判断客户喜不喜欢这个产品，或者是预测这个产品在下一个季度能卖多少出去。
　　这就涉及到机器学习的内容了。机器学习一般分为监督学习和非监督学习（supervised learning and unsupervised learning)。其中，监督学习又分为分类和回归，非监督学习分为聚类和降维。主要算法如下图
　　


Exploratory Data Analysis
　　我们一般从两个方面来研究数据：数据可视化和总结数据。
　　总结数据有个经典的方法：5-number summary。这个方法是要研究数据的5个数字：min，max，median，first quartiles，third quartiles。
　　一般在处理数据的时候，会进行可视化，主要包括：box plots，stem，leaf diagrams等。
　　R语言在处理这两个问题上，非常犀利。

Spark's Machine Learning Toolkit
　　Spark Mllib提供了非常强大的工具，如下图