正文

【原】Spark之机器学习(Python版)(一)——聚类

IP云
IP云V管理员/4 阅读
07/27
提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

  kmeans聚类相信大家都已经很熟悉了。在Python里我们用kmeans通常调用Sklearn包（当然自己写也很简单）。那么在Spark里能不能也直接使用sklean包呢？目前来说直接使用有点困难，不过我看到spark-packages里已经有了，但还没有发布。不过没关系，PySpark里有ml包，除了ml包，还可以使用MLlib，这个在后期会写，也很方便。

首先来看一下Spark自带的例子：

from pyspark.mllib.linalg import Vectors
from pyspark.ml.clustering import KMeans
from pyspark.sql import SQLContext
from pyspark.mllib.linalg import Vectors
#导入数据
data = [(Vectors.dense([0.0, 0.0]),), (Vectors.dense([1.0, 1.0]),),(Vectors.dense([9.0, 8.0]),), (Vectors.dense([8.0, 9.0]),)]
df = sqlContext.createDataFrame(data, ["features"])
#kmeans模型
kmeans = KMeans(k=2, seed=1)
model = kmeans.fit(df)
#簇心数量
centers = model.clusterCenters()
len(centers)
#
#训练模型
transformed = model.transform(df).select("features", "prediction")
rows = transformed.collect()
rows[0].prediction == rows[1].prediction
#True
rows[2].prediction == rows[3].prediction
# True

这个例子很简单，导入的数据是四个稠密向量(可以自己在二维向量里画一下)，设定了两个簇心，最后验证预测的结果是否正确，显示为True，证明预测正确。算法中具体的参数可以参考API中的说明。然而实际生产中我们的数据集不可能以这样的方式一条条写进去，一般是读取文件，关于怎么读取文件，可以具体看我的这篇博文。这里我们采用iris数据集（不要问我为什么又是iris数据集，因为真的太方便了）来给大家讲解一下。

我的数据集是csv格式的，而Spark又不能直接读取csv格式的数据，这里我们有两个方式，一是我提到的这篇博文里有写怎么读取csv文件，二是安装spark-csv包（在这里下载）,github地址在这里。按照步骤安装可以了。这里友情提示一下大家，github的安装方法是：

$SPARK_HOME/bin/spark-shell --packages com.databricks:spark-csv_2.11:1.4.0

如果报错了，可以把 --packages 换成 --jars，如果还是不行，在加一个 common-csv.jars包放到lib下面就可以了。我因为这个耽误了不少时间，不过具体问题也得具体分析。

安装好这个包以后，就可以读取数据了

from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)
data = sqlContext.read.format('com.databricks.spark.csv').options(header='true', inferschema='true').load('iris.csv')
data.show()

读取数据以后，我们来看一下数据集：

+------+------------+-----------+------------+-----------+-------+
|row.id|Sepal.Length|Sepal.Width|Petal.Length|Petal.Width|Species|
+------+------------+-----------+------------+-----------+-------+
| 1| 5.1| 3.5| 1.4| 0.2| 0|
| 2| 4.9| 3.0| 1.4| 0.2| 0|
| 3| 4.7| 3.2| 1.3| 0.2| 0|
| 4| 4.6| 3.1| 1.5| 0.2| 0|
| 5| 5.0| 3.6| 1.4| 0.2| 0|
| 6| 5.4| 3.9| 1.7| 0.4| 0|
| 7| 4.6| 3.4| 1.4| 0.3| 0|
| 8| 5.0| 3.4| 1.5| 0.2| 0|
| 9| 4.4| 2.9| 1.4| 0.2| 0|
| 10| 4.9| 3.1| 1.5| 0.1| 0|
| 11| 5.4| 3.7| 1.5| 0.2| 0|
| 12| 4.8| 3.4| 1.6| 0.2| 0|
| 13| 4.8| 3.0| 1.4| 0.1| 0|
| 14| 4.3| 3.0| 1.1| 0.1| 0|
| 15| 5.8| 4.0| 1.2| 0.2| 0|
| 16| 5.7| 4.4| 1.5| 0.4| 0|
| 17| 5.4| 3.9| 1.3| 0.4| 0|
| 18| 5.1| 3.5| 1.4| 0.3| 0|
| 19| 5.7| 3.8| 1.7| 0.3| 0|
| 20| 5.1| 3.8| 1.5| 0.3| 0|
+------+------------+-----------+------------+-----------+-------+
only showing top 20 rows

第二步：提取特征

我们在上一步导入的数据中label是String类型的，但在Spark中要变成数值型才能计算，不然就会报错。可以利用StringIndexer功能将字符串转化为数值型

from pyspark.ml.feature import StringIndexer feature = StringIndexer(inputCol="Species", outputCol="targetlabel")
target = feature.fit(data).transform(data)
target.show()

targetlabel这一列就是Species转化成数值型的结果

+------+------------+-----------+------------+-----------+-------+-----------+
|row.id|Sepal.Length|Sepal.Width|Petal.Length|Petal.Width|Species|targetlabel|
+------+------------+-----------+------------+-----------+-------+-----------+
| 1| 5.1| 3.5| 1.4| 0.2| 0| 0.0|
| 2| 4.9| 3.0| 1.4| 0.2| 0| 0.0|
| 3| 4.7| 3.2| 1.3| 0.2| 0| 0.0|
| 4| 4.6| 3.1| 1.5| 0.2| 0| 0.0|
| 5| 5.0| 3.6| 1.4| 0.2| 0| 0.0|
| 6| 5.4| 3.9| 1.7| 0.4| 0| 0.0|
| 7| 4.6| 3.4| 1.4| 0.3| 0| 0.0|
| 8| 5.0| 3.4| 1.5| 0.2| 0| 0.0|
| 9| 4.4| 2.9| 1.4| 0.2| 0| 0.0|
| 10| 4.9| 3.1| 1.5| 0.1| 0| 0.0|
| 11| 5.4| 3.7| 1.5| 0.2| 0| 0.0|
| 12| 4.8| 3.4| 1.6| 0.2| 0| 0.0|
| 13| 4.8| 3.0| 1.4| 0.1| 0| 0.0|
| 14| 4.3| 3.0| 1.1| 0.1| 0| 0.0|
| 15| 5.8| 4.0| 1.2| 0.2| 0| 0.0|
| 16| 5.7| 4.4| 1.5| 0.4| 0| 0.0|
| 17| 5.4| 3.9| 1.3| 0.4| 0| 0.0|
| 18| 5.1| 3.5| 1.4| 0.3| 0| 0.0|
| 19| 5.7| 3.8| 1.7| 0.3| 0| 0.0|
| 20| 5.1| 3.8| 1.5| 0.3| 0| 0.0|
+------+------------+-----------+------------+-----------+-------+-----------+
only showing top 20 rows

最后一步：模型训练和验证

from pyspark.sql import Row
from pyspark.ml.clustering import KMeans
from pyspark.mllib.linalg import Vectors #把数据格式转化成稠密向量
def transData(row):
return Row(label=row["targetlabel"],
features=Vectors.dense([row["Sepal.Length"],
row["Sepal.Width"],
row["Petal.Length"],
row["Petal.Width"]])) #转化成Dataframe格式
transformed = target.map(transData).toDF()
kmeans = KMeans(k=3)
model = kmeans.fit(transformed) predict_data = model.transform(transformed) train_err = predict_data.filter(predict_data['label'] != predict_data['prediction']).count()
total = predict_data.count()
print traing_err, total, float(train_err)/total

到这一步就结束了。总结一下，用pyspark做机器学习时，数据格式要转成需要的格式，不然很容易出错。下周写pyspark在机器学习中如何做分类。