问题导读
1、你如何理解转换(Transformations)的含义?
2、Spark实现机器学习算法的流程有哪些?
3、如何利用已有的训练样本对模型进行训练?
分布式数据集支持两种操作:
1. 转换(transformation):根据现有的数据集创建一个新的数据集
2. 动作(actions):在数据集上运行计算后,返回一个值给驱动程序
所有Spark中的转换都是惰性的,也就是说,并不会马上发生计算。相反的,它只是记住应用到基础数据集上的这些转换(Transformation)。而这些转换(Transformation),只会在有一个动作(Action)发生,要求返回结果给驱动应用时,才真正进行计算。这个设计让Spark更加有效率的运行。例如,我们可以实现,通过map创建一个数据集,然后再用reduce,而只返回reduce的结果给driver,而不是整个大的数据集。
转换(Transformations)
Transformation | | | 返回一个新的分布式数据集,由每个原元素经过func函数转换后组成 | | 返回一个新的数据集,由经过func函数后返回值为true的原元素组成 | | 类似于map,但是每一个输入元素,会被映射为0到多个输出元素(因此,func函数的返回值是一个Seq,而不是单一元素) | sample(withReplacement, frac, seed) | 根据给定的随机种子seed,随机抽样出数量为frac的数据 | | | | 在一个由(K,V)对组成的数据集上调用,返回一个(K,Seq[V])对的数据集。注意:默认情况下,使用8个并行任务进行分组,你可以传入numTask可选参数,根据数据量设置不同数目的Task(groupByKey和filter结合,可以实现类似Hadoop中的Reduce功能) | reduceByKey(func, [numTasks]) | 在一个(K,V)对的数据集上使用,返回一个(K,V)对的数据集,key相同的值,都被使用指定的reduce函数聚合到一起。和groupbykey类似,任务的个数是可以通过第二个可选参数来配置的。 | join(otherDataset, [numTasks]) | 在类型为(K,V)和(K,W)类型的数据集上调用,返回一个(K,(V,W))对,每个key中的所有元素都在一起的数据集 | groupWith(otherDataset, [numTasks]) | 在类型为(K,V)和(K,W)类型的数据集上调用,返回一个数据集,组成元素为(K, Seq[V], Seq[W]) Tuples。这个操作在其它框架,称为CoGroup | | 笛卡尔积。但在数据集T和U上调用时,返回一个(T,U)对的数据集,所有元素交互进行笛卡尔积。 | sortByKey([ascendingOrder]) | 在类型为( K, V )的数据集上调用,返回以K为键进行排序的(K,V)对数据集。升序或者降序由boolean型的ascendingOrder参数决定(类似于Hadoop的Map-Reduce中间阶段的Sort,按Key进行排序) |
Actions(动作) Action | | | 通过函数func聚集数据集中的所有元素。Func函数接受2个参数,返回一个值。这个函数必须是关联性的,确保可以被正确的并发执行 | | 在Driver的程序中,以数组的形式,返回数据集的所有元素。这通常会在使用filter或者其它操作后,返回一个足够小的数据子集再使用,直接将整个RDD集Collect返回,很可能会让Driver程序OOM | | | | 返回一个数组,由数据集的前n个元素组成。注意,这个操作目前并非在多个节点上,并行执行,而是Driver程序所在机器,单机计算所有的元素(Gateway的内存压力会增大,需要谨慎使用) | | | | 将数据集的元素,以textfile的形式,保存到本地文件系统,hdfs或者任何其它hadoop支持的文件系统。Spark将会调用每个元素的toString方法,并将它转换为文件中的一行文本 | | 将数据集的元素,以sequencefile的格式,保存到指定的目录下,本地系统,hdfs或者任何其它hadoop支持的文件系统。RDD的元素必须由key-value对组成,并都实现了Hadoop的Writable接口,或隐式可以转换为Writable(Spark包括了基本类型的转换,例如Int,Double,String等等) | | 在数据集的每一个元素上,运行函数func。这通常用于更新一个累加器变量,或者和外部存储系统做交互 |
Spark MLlib系列——程序框架
本系列将对Spark中的pyspark的mllib模块进行分析。这篇文章先简要介绍一下Spark实现机器学习算法的流程。
大部分的机器学习算法都包含训练以及预测两个部分,训练出模型,然后对未知样本进行预测。Spark中的机器学习包完全体现了这样一种思想,下面我们看一个最简单的例子:
- class LinearModel(object):
- """A linear model that has a vector of coefficients and an intercept."""
- def __init__(self, weights, intercept):
- self._coeff = weights
- self._intercept = intercept
-
- @property
- def weights(self):
- return self._coeff
-
- @property
- def intercept(self):
- return self._intercept
-
- class LinearRegressionModelBase(LinearModel):
- """A linear regression model.
-
- >>> lrmb = LinearRegressionModelBase(array([1.0, 2.0]), 0.1)
- >>> abs(lrmb.predict(array([-1.03, 7.777])) - 14.624) < 1e-6
- True
- >>> abs(lrmb.predict(SparseVector(2, {0: -1.03, 1: 7.777})) - 14.624) < 1e-6
- True
- """
- def predict(self, x):
- """Predict the value of the dependent variable given a vector x"""
- """containing values for the independent variables."""
- _linear_predictor_typecheck(x, self._coeff)
- return _dot(x, self._coeff) + self._intercept
复制代码
这是Spark MLlib的线性回归的预测模型,回归模型的两个参数,一个是特征向量对应的权重向量(weights),另一个是截距(intercept),而预测样本的回归值则只需要输入样本的特征向量。因此,在线性回归模型中,用模型参数——权重向量以及截距,来初始化,预测函数predict输入待测样本的特征向量,并输出预测值。
现在的关键是如何得到模型参数,这就需要利用已有的训练样本对模型进行训练。
- class LinearRegressionWithSGD(object):
- @classmethod
- def train(cls, data, iterations=100, step=1.0,
- miniBatchFraction=1.0, initialWeights=None):
- """Train a linear regression model on the given data."""
- sc = data.context
- train_f = lambda d, i: sc._jvm.PythonMLLibAPI().trainLinearRegressionModelWithSGD(
- d._jrdd, iterations, step, miniBatchFraction, i)
- return _regression_train_wrapper(sc, train_f, LinearRegressionModel, data, initialWeights)
-
- def _regression_train_wrapper(sc, train_func, klass, data, initial_weights):
- initial_weights = _get_initial_weights(initial_weights, data)
- dataBytes = _get_unmangled_labeled_point_rdd(data)
- ans = train_func(dataBytes, _serialize_double_vector(initial_weights))
- if len(ans) != 2:
- raise RuntimeError("JVM call result had unexpected length")
- elif type(ans[0]) != bytearray:
- raise RuntimeError("JVM call result had first element of type "
- + type(ans[0]).__name__ + " which is not bytearray")
- elif type(ans[1]) != float:
- raise RuntimeError("JVM call result had second element of type "
- + type(ans[0]).__name__ + " which is not float")
- return klass(_deserialize_double_vector(ans[0]), ans[1])
复制代码
由上面的代码可以看到,线性回归的训练调用的是封装好的trainLinearRegressionModelWithSGD函数,其输出值ans有两个,一个是weights,另一个是intercept。_regression_train_wrapper函数的作用是对训练模型的参数进行包装,并返回预测模型对象。
通过上面对线性回归模型代码分析可以知道,Spark将机器学习算法都分成了两个模块:一个训练模块,通过训练样本输出模型参数;另一个是预测模块,利用模型参数初始化,预测测试样本,输出与测值。
其他的机器学习算法都类似,当然有些机器学习算法天生不用经过训练的过程,如KNN,自然也就没有训练阶段。
|