Spark 数据ETL及部分代码示例

问题导读：




1.数据如何处理？
2.从数据中如何提取有用的特征？
3.有哪些衍生特征？








数据处理以及转化

1、当我们完成了一些对数据集的探索和分析，我们知道了一些关于用户数据以及电影数据的特征，接下来我们该做些什么呢？

2、为了让原始数据能够在机器学习算法中变得有用，我们首先需要清理以及在提取有用的特征值之前使用各种方法尽可能地转化它。其中的转化和特征提取步骤是紧密连接的，而且在一些情况下，特定的转化就是一种特征值提取的过程。

3、我们已经看过了在电影数据集中需要清理数据的例子。通常，现实的数据集包含坏的数据、丢失的数据以及异常值。理想情况下，我们可以纠正错误的数据；但是，这通常都是不可能的。因为很多数据集来源于那些不能够重复的集合操作。丢失的数据以及异常值也是很常见的，它们可以用类似于坏数据的处理方法处理。总的来说，归结为以下广泛的处理方法：

过滤掉或者移除坏数据以及丢失的数据：

有时候这是不可避免的；然而这也意味着丢失掉大部分坏的或丢失的记录。

填充坏掉或者丢失的数据：

我们可以尽力地依据剩下的数据来给坏掉的或者丢失的数据赋值。比如赋给0值、平均值、中位数、附近的值或者相似值等方法。选择正确的方法通常是一件棘手的任务,这取决于数据、情况和自己的经验。

应用成熟的技术到异常值：

异常值的主要问题在于它们的值可能是正确的，尽管它们是极端值。它们也有可能是错误的。所以很难知道我们处理的是哪种情况。异常值也可以被移除或者填充。不过幸运的是,有是统计技术(如稳健回归)来处理异常值和极端值。

转化潜在的异常值：

另一个处理异常值或者极端值得方法是转化。例如对数或者高斯内核转化，计算出潜在的异常值，或者显示大范围的潜在数据。这些类型的转换抑制了变量大尺度变化的影响并将非线性关系转化为一个线性的。

填充坏的或丢失的数据：

我们之前已经见过过滤坏数据的例子了。我们接着之前的代码，下面的代码段对坏数据应用了填充的方法，通过赋给数据点以相等于year中值的值。

[mw_shl_code=python,true]
years_pre_processed = movie_fields.map(lambda fields: fields[2]).  
map(lambda x: convert_year(x)).collect()  
years_pre_processed_array = np.array(years_pre_processed)  
[/mw_shl_code]

首先，我们将在选择所有的发布年限后计算year的平均值和中位数，除了那些坏的数据。之后使用numpy函数，从years_pre_processed_array中查找坏数据的索引（参考之前我们赋予1900给数据点）。最后，我们使用这个索引来赋予中值给坏的数据：

[mw_shl_code=python,true]
mean_year = np.mean(years_pre_processed_array[years_pre_processed_  
array!=1900])  
median_year = np.median(years_pre_processed_array[years_pre_processed_  
array!=1900])  
index_bad_data = np.where(years_pre_processed_array==1900)[0][0]  
years_pre_processed_array[index_bad_data] = median_year  
print "Mean year of release: %d" % mean_year  
print "Median year of release: %d" % median_year  
print "Index of '1900' after assigning median: %s" % np.where(years_  
pre_processed_array == 1900)[0][/mw_shl_code]

打印结果应该类似于如下：

[mw_shl_code=python,true]Mean year of release: 1989  
Median year of release: 1995  
Index of '1900' after assigning median: []  [/mw_shl_code]

在这里我们计算出year的平均值和中位数，从输出结果中我们可以看出，year的中位数因为year的倾斜分布要比平均值高许多。尽管直接决定使用一个精确的值去填充数据不是常见的做法，但是由于数据的倾斜，使用中位数去赋值是一种可行的方法。

从数据中提取有用的特征

1、当我们完成了对数据初始的处理和清洗，我们就可以准备从数据中提取一些实际有用的特征，这些特征数据可以用于以后的机器学习模型中的训练。

2、特征数据是指我们用于训练模型的一些变量。每行数据都有可能包含可以提取用于训练的样例。几乎所有的机器学习模型都是工作在以数字为技术的向量数据上。因此，我们需要将粗糙的数据转化为数字。

特征数据可以分为以下几类：

数字特征

这类特征数据是指一些数值类型的数据。

分类特征

这类特征数据代表一些相同特性的，可以归为一类的一些数据。例如用户的性别、职位或者电影的类型。

文本特征

这类特征数据是从数据中的文本内容中派生出来的，例如电影名称，描述，以及评论。

其他特征

这类特征数据都会转化为以数字为代表的特征，例如图片，视频，音频都可以表示为数字数据的集合。地理位置可以代表为经度、纬度或者经纬度之差。

数字特征

1、旧数字和提取的新的特征数值有什么区别呢？其实，在现实生活中，任何的数值数据都可以作为输入变量，但在机器学习模型中，我们学习的是每个特征的向量权重，例如监督学习模型。

2、因此，我们需要使用那些有意义的特征数据，那些模型可以从特征值与目标数据之间学习关系的特征数据。例如，年龄就是一个合理的特征数据，比如年龄的增长和产出有着直接的关系，同样，身高也是可以直接使用的数值特征。

分类特征

1、分类特征数据不能直接使用它们原有的粗糙的格式作为输入使用，因为它们不是数字。但是它们其中的一些衍生值可以作为输入的变量。比如之前所说的职位就可以有学生、程序员等。

2、这些分类变量只是名义上的变量，因为它们不存在变量值之间的顺序的概念。相反，当变量之间存顺序概念时，我们会倾向于使用这些常见有序的变量。

3、为了把这些分类变量转化为数字表示，我们可以使用常用的方法，例如1-of-k编码。这种方法需要把那些名义上的变量转化为对机器学习任务有用的数据。常见那些粗糙格式的数据都会以名义上的变量形式编码为有意义的数据。

4、我们假设这里有k个值可以供变量获取，如果我们可以给每个值都赋予1到k中的索引，然后我们就可以使用程度为k的二进制向量表示一个值了。初始的实体中，向量表示的二进制值都是0，当我们赋予变量一个状态的时候，所对应的二进制向量中对应的索引值由0变成1。

例如，我们先获取上面所说的职位的所有类别变量：

[mw_shl_code=python,true]all_occupations = user_fields.map(lambda fields: fields[3]).  
distinct().collect()  
all_occupations.sort()  [/mw_shl_code]

接着我们可以赋给每个可能的职位类别一个值（值得索引从零开始，因为在Python、Scala、Java数组中索引都是从0开始的）

[mw_shl_code=python,true]
    idx = 0  
    all_occupations_dict = {}  
    for o in all_occupations:  
        all_occupations_dict[o] = idx  
    idx += 1  
    # try a few examples to see what "1-of-k" encoding is assigned  
    print "Encoding of 'doctor': %d" % all_occupations_dict['doctor']  
    print "Encoding of 'programmer': %d" % all_occupations_  
    dict['programmer'] [/mw_shl_code]

你将看到如下打印结果：

[mw_shl_code=python,true]
    Encoding of 'doctor': 2  
    Encoding of 'programmer': 14  [/mw_shl_code]

最后我们可以对上面打印的结果中programmer进行编码，我们可以首先创建一个长度为k（在这个案例中）的numpy数组并且值全部填0（我们将使用numpy数组中的zeros函数创建这个数组）。

我们将提取单词programmer的索引并赋予1给数组的这个索引：

[mw_shl_code=python,true]
    K = len(all_occupations_dict)  
    binary_x = np.zeros(K)  
    k_programmer = all_occupations_dict['programmer']  
    binary_x[k_programmer] = 1  
    print "Binary feature vector: %s" % binary_x  
    print "Length of binary vector: %d" % K  [/mw_shl_code]

上面结果将呈现给我们长度为21的二进制特征的向量：

[mw_shl_code=python,true]
    Binary feature vector: [ 0. 0. 0. 0. 0. 0. 0. 0. 0. 0. 0. 0.  
                             0. 0. 1. 0. 0. 0. 0. 0. 0.]  
    Length of binary vector: 21 [/mw_shl_code]

衍生特征

1、通常会从一或多个可获得的变量中计算出衍生特征是很有用的，我们希望那些衍生特征可以相比于原来粗糙格式的变量添加更多的信息。

2、例如，我们可以计算所有电影评分数据中的用户平均评分，用户平均评分将提供针对用户截差的模型。我们已经获取了粗糙的评分数据，并且创建了新的可以让我们学习更好模型的特征。

3、从粗糙数据中获取衍生特征数据的例子包括平均值、中位数、求和、最大值、最小值以及总数等。比如在电影数据中，我们可以通过现在的年限减去电影发布年限获得电影的年龄。

4、通常，这些转化用来产生数值数据以便于更好的让模型去学习。

5、把数字特征值转化为分类特征值也很常见，比如

转化timestamps值为分类特征值

为了演示怎样从数字特征值衍生为分类特征值，我们将使用电影评分数据中的评分时间。这些时间都是Unix timestamps格式。我们可以用Python的datetime模块去从timestamp中获取date和time，然后提取day中的hour。这将为每个评分中day的hour成一个RDD。

我们将需要一个函数去提取代表评分timestamp的datetime：

[mw_shl_code=python,true]
    def extract_datetime(ts):  
        import datetime  
        return datetime.datetime.fromtimestamp(ts)  [/mw_shl_code]

我们继续使用之前例子之中计算出的rating_data RDD

首先，我们使用map转化提取timestamp列，把它转化为Python中int类型。对每个timestamp应用extract_datetime方法，然后从结果datetime对象中提取hour:
[mw_shl_code=python,true]timestamps = rating_data.map(lambda fields: int(fields[3]))  
hour_of_day = timestamps.map(lambda ts: extract_datetime(ts).hour)  
hour_of_day.take(5)  [/mw_shl_code]

如果我们从结果RDD中获取前五条记录，我们将看到以下输出结果：

[mw_shl_code=python,true][17, 21, 9, 7, 7]  [/mw_shl_code]

至此我们已经将粗糙的时间数据转化为了评分数据中代表day中hour的分类特征数据

现在，我们说的这种转化可能优点粗糙，也许我们想更加贴切地定义转化。我们可以将每天中的小时转化为代表每天时间中的块。例如我们可以定义morning是从7 am到 11 am、lunch是从11 am到 1am等。使用这些块，我们可以创建方法给每天中的时间赋值，下面将day中的hour作为输入：
[mw_shl_code=python,true]def assign_tod(hr):  
    times_of_day = {  
        'morning' : range(7, 12),  
        'lunch' : range(12, 14),  
        'afternoon' : range(14, 18),  
        'evening' : range(18, 23),  
        'night' : range(23, 7)  
    }  
for k, v in times_of_day.iteritems():  
    if hr in v:  
    return k  [/mw_shl_code]

现在，我们可以将assign_tod函数应用到存在于hour_of_day RDD中的每个评分记录中的hour上。

[mw_shl_code=python,true]
    time_of_day = hour_of_day.map(lambda hr: assign_tod(hr))  
    time_of_day.take(5)  [/mw_shl_code]

如果我们获取这个RDD的前5条记录，我们将看到如下转化后的值：

[mw_shl_code=python,true]    ['afternoon', 'evening', 'morning', 'morning', 'morning']  [/mw_shl_code]

到此，我们已经将timestamp变量转化为24小时格式的hours变量，以及自定义的每天中的时间值。因此我们已经有了分类特征值，可以使用之前介绍的1-of-k编码方法去生成二进制特征的向量。

文本特征值

1、在某些情况下，文本特征值是以分类以及衍生特征存在的。我们拿电影的描述信息作为例子。这里，粗糙的数据不能被直接使用，即使是作为分类特征，因为如果每个文本都有值，那将会产生无限种可能组合的单词。我们的模型几乎不会出现两种相同特征，就算有那么学习效率也不会高。因此,我们希望将原始文本变成一种更适合机器学习的形式。

2、有很多的方法可以处理文本，而且自然语言领域处理致力于处理、呈现和模型化文本内容。我们将介绍简单和标准的方法来实现文本特征提取，这个方法就是词袋模型表示。

3、词袋模型将文本块视为单词的集合以及可能存在的数字，词袋方法的处理如下：

标记：首先，一些形式的标记用于将文本分割为标记的集合（一般是单词，数字等）。例如常见的空格标记，将文本按照每个空格分隔，还有其他的一些标点和非字母数字的标记。

可以移除的停止词：一般我们会移除文本中非常常见的词，例如”the”、”and”、”but”（这些都称为停止词）。

词干提取：接下来的操作包括词干提取，一种获取输入项，然后将其提取为其最基础的值。一个常见的例子就是复数编程单数，或者dogs变成dog。有很多方法可以实现词干提取，有很多文本处理库也包含各种词干提取算法。

向量化：最后一步是将处理项转化为向量表示形式。最简单的形式也许就是二进制的向量表示形式，如果一个处理项包含在文本中，我们就给它赋值为1，如果没有就赋值为0。本质上是我们之前提到的分类的1-of-k编码。类似1-of-k编码，这里需要一个字典将这些项映射为一个个索引。也许你会想到，这里可能存在几百万单独项。因此，使用稀疏向量表示形式是非常严格的，只在那些处理项已被保存的情况下使用，这样可以节省内存、磁盘空间以及处理时间。

简单文本特征提取

我们使用电影评分数据中的电影名称演示以二进制向量方法提取文本特征值。

首先我们创建函数去除每部电影的发布年限，仅留下电影名称。

电影数据示例：

[mw_shl_code=python,true]1|Toy Story (1995)|01-Jan-1995||http://us.imdb.com/M/title-exact?Toy%20Story%20(1995)|0|0|0|1|1|1|0|0|0|0|0|0|0|0|0|0|0|0|0  [/mw_shl_code]

我们将使用Python的 regular expression模块re，去搜索出存在于电影名称列的电影发布年限。当我们匹配到这个regular expression，我们将只提取出电影名称，示例：

[mw_shl_code=python,true]    def extract_title(raw):  
        import re  
        # this regular expression finds the non-word (numbers) between  
        parentheses  
        grps = re.search("(\w+)", raw)  
        if grps:  
        # we take only the title part, and strip the trailingwhite spacefrom the remaining text, below  
            return raw[:grps.start()].strip()  
        else:  
            return raw  [/mw_shl_code]

接下来，我们将从movie_fields RDD中提取出粗糙的电影名称：

[mw_shl_code=python,true]
    //包含电影发布年限，格式：Toy Story (1995)  
    raw_titles = movie_fields.map(lambda fields: fields[1])  [/mw_shl_code]

然后我们通过下面的代码提取5条记录测试extract_title函数的功能：

[mw_shl_code=python,true]    for raw_title in raw_titles.take(5):  
        print extract_title(raw_title)  [/mw_shl_code]

通过打印结果我们可以验证函数执行情况，打印结果示例：

[mw_shl_code=python,true]Toy Story  
GoldenEye  
Four Rooms  
Get Shorty  
Copycat  [/mw_shl_code]

我们将应用函数以及标记模式来提取电影名称为单个元素，下面我们使用简单地空格标记来分离电影名称。

[mw_shl_code=python,true]movie_titles = raw_titles.map(lambda m: extract_title(m))  
# next we tokenize the titles into terms. We'll use simple whitespace  
tokenization  
title_terms = movie_titles.map(lambda t: t.split(" "))  
print title_terms.take(5)  [/mw_shl_code]

打印结果：

[mw_shl_code=python,true][[u'Toy', u'Story'], [u'GoldenEye'], [u'Four', u'Rooms'], [u'Get',u'Shorty'], [u'Copycat']]  [/mw_shl_code]

现在我们可以看出电影名称以及被按照空格分离为单个的标记了。

为了给每一项赋值一个向量的索引,我们需要创建词典,将每一项都映射到一个整数索引。

首先，我们将使用Spark的flatMap函数来扩张title_terms RDD中每条记录的list字符串，转化为每条记录都是一项的名为all_terms的RDD。

我们获取所有的唯一项，然后赋值索引，就像之前的对职位操作的1-of-k编码。

[mw_shl_code=python,true]# next we would like to collect all the possible terms, in order to  
build out dictionary of term <-> index mappings  
all_terms = title_terms.flatMap(lambda x: x).distinct().collect()  
# create a new dictionary to hold the terms, and assign the "1-of-k"  
indexes  
idx = 0  
all_terms_dict = {}  
for term in all_terms:  
    all_terms_dict[term] = idx  
idx +=1  [/mw_shl_code]

我们打印出唯一项的总数来测试我们的map功能是否正常工作：

[mw_shl_code=python,true]print "Total number of terms: %d" % len(all_terms_dict)  
print "Index of term 'Dead': %d" % all_terms_dict['Dead']  
print "Index of term 'Rooms': %d" % all_terms_dict['Rooms']  [/mw_shl_code]

打印结果：

[mw_shl_code=python,true]Total number of terms: 2645  
Index of term 'Dead': 147  
Index of term 'Rooms': 1963  [/mw_shl_code]

我们也可以使用Spark的zipWithIndex函数来更加有效地实现上面的结果，这个函数获取values的RDD然后通过索引合并它们并且创建一个新的key-value对RDD，这个新的RDD的key就是唯一项，value是这个项的字典索引。我们通过使用collectAsMap函数来将这个key-value RDD作为Python字典方法传入driver。

[mw_shl_code=python,true]all_terms_dict2 = title_terms.flatMap(lambda x: x).distinct().  
zipWithIndex().collectAsMap()  
print "Index of term 'Dead': %d" % all_terms_dict2['Dead']  
print "Index of term 'Rooms': %d" % all_terms_dict2['Rooms']  [/mw_shl_code]

打印结果：

[mw_shl_code=python,true]
    Index of term 'Dead': 147  
    Index of term 'Rooms': 1963  [/mw_shl_code]

最后一步是创建一个函数将唯一项的集合转化为一个稀疏的向量表示形式。为了达到效果，我们将创建一个空的，有一行以及和字典中唯一项总数的列的稀疏矩阵。然后我们将通过输入列表中的每一项来检查这一项是否存在于我们的唯一项字典中。如果是，我们将给这个字典中对应的这个唯一项的索引赋值为1。

[mw_shl_code=python,true]# this function takes a list of terms and encodes it as a scipy sparse  
vector using an approach  
# similar to the 1-of-k encoding  
def create_vector(terms, term_dict):  
    from scipy import sparse as sp  
        num_terms = len(term_dict)  
        x = sp.csc_matrix((1, num_terms))  
        for t in terms:  
            if t in term_dict:  
                idx = term_dict[t]  
                x[0, idx] = 1  
    return x  [/mw_shl_code]

当我们有了上面的函数之后，我们会将它应用到提取项的RDD中的每一条记录中。

[mw_shl_code=python,true]all_terms_bcast = sc.broadcast(all_terms_dict)  
term_vectors = title_terms.map(lambda terms: create_vector(terms, all_  
terms_bcast.value))  
term_vectors.take(5)  [/mw_shl_code]

我们可以查看一些稀疏向量新的RDD的执行记录：

[mw_shl_code=python,true][<1x2645 sparse matrix of type '<type 'numpy.float64'>'  
with 2 stored elements in Compressed Sparse Column format>,  
<1x2645 sparse matrix of type '<type 'numpy.float64'>'  
with 1 stored elements in Compressed Sparse Column format>,  
<1x2645 sparse matrix of type '<type 'numpy.float64'>'  
with 2 stored elements in Compressed Sparse Column format>,  
<1x2645 sparse matrix of type '<type 'numpy.float64'>'  
with 2 stored elements in Compressed Sparse Column format>,  
<1x2645 sparse matrix of type '<type 'numpy.float64'>'  
with 1 stored elements in Compressed Sparse Column format>]  [/mw_shl_code]

从上面的记录中可以看到电影名称已经被转化为一个稀疏向量，我们可以看出那些提取为两个单词的电影名称项在向量中有两个非0的实体，那些提取为一个单词名称项的在向量中有一个非0实体，其他类似。

标准化特征值

一旦特征值被提取为向量的形式，通常的预处理步骤为标准化数值数据。思路为通过将每个值转化为标准大小的方法转化每一个数值特征值。我们有不同的方法来标准化，如下：

标准化一个特征：

这通常应用于数据集中的一个单独的特征，例如减去均值（使特征值中心化）或者使用标准正态变换。

标准化特征向量：

这通常是应用转化到数据集中给定一行的所有特征值，这样得到的特征向量有一个标准化的长度。也就是说,我们将确保每个特征向量是以1为基础按比例缩小的。

下面我们使用第二种方法作为例子。我们将使用Python中numpy模块的norm函数来实现向量的标准化，,首先计算一个L2规范的随机向量，然后使用这种规范分离向量中每个元素除来创建我们标准化的向量。

[mw_shl_code=python,true]np.random.seed(42)  
x = np.random.randn(10)  
norm_x_2 = np.linalg.norm(x)  
normalized_x = x / norm_x_2  
print "x:\n%s" % x  
print "2-Norm of x: %2.4f" % norm_x_2  
print "Normalized x:\n%s" % normalized_x  
print "2-Norm of normalized_x: %2.4f" %  
np.linalg.norm(normalized_x)  [/mw_shl_code]

上面将给出如下结果，

[mw_shl_code=python,true]x: [ 0.49671415 -0.1382643 0.64768854 1.52302986 -0.23415337 -0.23413696  
     1.57921282 0.76743473 -0.46947439 0.54256004]  
2-Norm of x: 2.5908  
Normalized x: [ 0.19172213 -0.05336737 0.24999534 0.58786029 -0.09037871  
                -0.09037237 0.60954584 0.29621508 -0.1812081 0.20941776]  
2-Norm of normalized_x: 1.0000  [/mw_shl_code]



使用MLlib实现特征标准化

Spark在它的MLlib机器学习库中内建了一些功能扩展以及标准化的函数。包括StandardScaler，用于标准正态变换；以及Normalizer，提供了我们之前处理示例代码中的向量标准化功能。

下面让我们简单地使用MLlib的Normalizer来比较之前的结果：

[mw_shl_code=python,true]from pyspark.mllib.feature import Normalizer  
normalizer = Normalizer()  
vector = sc.parallelize([x])  [/mw_shl_code]

当引入需要的类以后，我们将实例化Normalizer。注意，在大多数的Spark解决方案中，我们需要带RDD的Normalizer作为输入（包含了numpy数组或者MLlib向量）；因此，我们将从向量x创建单元素的RDD来作为演示。

我们将在RDD上使用Normalizer的transform函数。当RDD中最终只有一个向量时，我们通过调用first来返回向量到driver中，最后调用toArray函数将vector转化回numpy数组。

[mw_shl_code=python,true]
    normalized_x_mllib = normalizer.transform(vector).first().toArray()  [/mw_shl_code]

最后，我们打印详情，和之前所做的一样，对比结果：

[mw_shl_code=python,true]    print "x:\n%s" % x  
    print "2-Norm of x: %2.4f" % norm_x_2  
    print "Normalized x MLlib:\n%s" % normalized_x_mllib  
    print "2-Norm of normalized_x_mllib: %2.4f" % np.linalg.  
    norm(normalized_x_mllib)  [/mw_shl_code]

你最后会发现通过我们的代码实现了同样的标准化向量。但是，使用MLlib中内建的方法比我们自己的函数要更加方便和高效。