如何将参数传递给 ML Pipeline.fit 方法? [英] How to pass params to a ML Pipeline.fit method?
问题描述
我正在尝试使用
- Google Dataproc + Spark
- Google Bigquery
- 使用 Spark ML KMeans+pipeline 创建作业
如下:
在 bigquery 中创建基于用户级别的特征表
示例:特征表的样子
Create user level based feature table in bigquery
Example: How the feature table looks like
<代码>用户 ID |x1 |x2 |x3 |x4 |x5 |x6 |x7 |x8 |x9 |x10
00013 |0.01 |0 |0 |0 |0 |0 |0 |0.06 |0.09 |0.001
- 启动默认设置集群，我使用 gcloud 命令行界面创建集群并运行作业，如图所示 这里
- 使用提供的入门代码，我读取 BQ 表，将 RDD 转换为数据帧并传递给 KMeans 模型/管道:
#!/usr/bin/python
"""BigQuery I/O PySpark example."""
import json
import pprint
import subprocess
import pyspark
import numpy as np
from pyspark.ml.clustering import KMeans
from pyspark import SparkContext
from pyspark.ml import Pipeline
from pyspark.sql import SQLContext
from pyspark.mllib.linalg import Vectors, _convert_to_vector
from pyspark.sql.types import Row
from pyspark.mllib.common import callMLlibFunc, callJavaFunc, _py2java, _java2py
sc = pyspark.SparkContext()
# Use the Google Cloud Storage bucket for temporary BigQuery export data used by the InputFormat.
# This assumes the Google Cloud Storage connector for Hadoop is configured.
bucket = sc._jsc.hadoopConfiguration().get('fs.gs.system.bucket')
project = sc._jsc.hadoopConfiguration().get('fs.gs.project.id')
input_directory ='gs://{}/hadoop/tmp/bigquery/pyspark_input'.format(bucket)
conf = {# Input Parameters
'mapred.bq.project.id': project,
'mapred.bq.gcs.bucket': bucket,
'mapred.bq.temp.gcs.path': input_directory,
'mapred.bq.input.project.id': 'my-project',
'mapred.bq.input.dataset.id': 'tempData',
'mapred.bq.input.table.id': 'userFeatureInBQ'}
# Load data in from BigQuery.
table_data = sc.newAPIHadoopRDD(
'com.google.cloud.hadoop.io.bigquery.JsonTextBigQueryInputFormat',
'org.apache.hadoop.io.LongWritable',
'com.google.gson.JsonObject',conf=conf)
# Tranform the userid-Feature table into feature_data RDD
feature_data = (
table_data
.map(lambda (_, record): json.loads(record))
.map(lambda x:(x['x0'],x['x1'],x['x2'],x['x3'],x['x4'],
x['x5'],x['x6'],x['x7'],x['x8'],
x['x9'],x['x10'])))
# Function to convert each line in RDD into an array, return the vector
def parseVector(values):
array = np.array([float(v) for v in values])
return _convert_to_vector(array)
# Convert the RDD into a row wise RDD
data = feature_data.map(parseVector)
row_rdd = data.map(lambda x: Row(x))
sqlContext = SQLContext(sc)
# cache the RDD to improve performance
row_rdd.cache()
# Create a Dataframe
df = sqlContext.createDataFrame(row_rdd, ["features"])
# cache the Dataframe
df.cache()
这是我打印到控制台的 Schema 和 head():
Here is the Schema and head() which I print to the console:
|-- features: vector (nullable = true)
[Row(features=DenseVector([0.01,0,0,0,0,0,0,0.06,0.09,0.001]))]
<小时>
- 按以下方式运行聚类 KMeans 算法
- 多次运行模型
- 使用不同的参数(即更改#clusters 和init_mode)
- 计算错误或成本指标
- 选择最佳模型参数组合
- 使用 KMeans 作为估计器创建管道
- 使用 paramMap 传递多个参数
#Define the paramMap & model
paramMap = ({'k':3,'initMode':'kmeans||'},{'k':3,'initMode':'random'},
{'k':4,'initMode':'kmeans||'},{'k':4,'initMode':'random'},
{'k':5,'initMode':'kmeans||'},{'k':5,'initMode':'random'},
{'k':6,'initMode':'kmeans||'},{'k':6,'initMode':'random'},
{'k':7,'initMode':'kmeans||'},{'k':7,'initMode':'random'},
{'k':8,'initMode':'kmeans||'},{'k':8,'initMode':'random'},
{'k':9,'initMode':'kmeans||'},{'k':9,'initMode':'random'},
{'k':10,'initMode':'kmeans||'},{'k':10,'initMode':'random'})
km = KMeans()
# Create a Pipeline with estimator stage
pipeline = Pipeline(stages=[km])
# Call & fit the pipeline with the paramMap
models = pipeline.fit(df, paramMap)`
print models
<小时>
我得到以下带有警告的输出
I get the following output with a warning
7:03:24 WARN org.apache.spark.mllib.clustering.KMeans:输入数据没有直接缓存，如果它的父 RDD 也没有缓存，这可能会影响性能.[PipelineModel_443dbf939b7bd3bf7bfc，PipelineModel_4b64bb761f4efe51da50，PipelineModel_4f858411ac19beacc1a4，PipelineModel_4f58b894f1d14d79b936，PipelineModel_4b8194f7a5e6be6eaf33，PipelineModel_4fc5b6370bff1b4d7dba，PipelineModel_43e0a196f16cfd3dae57，PipelineModel_47318a54000b6826b20e，PipelineModel_411bbe1c32db6bf0a92b，PipelineModel_421ea1364d8c4c9968c8，PipelineModel_4acf9cdbfda184b00328，PipelineModel_42d1a0c61c5e45cdb3cd，PipelineModel_4f0db3c394bcc2bb9352，PipelineModel_441697f2748328de251c，PipelineModel_4a64ae517d270a1e0d5a，PipelineModel_4372bc8db92b184c05b0]
#Print the cluster centers:
for model in models:
print vars(model)
print model.stages[0].clusterCenters()
print model.extractParamMap()
输出:<代码>[数组([7.64676638e-07, 3.58531391e-01, 1.68879698e-03, 0.00000000e+00, 1.53477043e-02, 1.25820206e-04,06e-06e-06,706e-06,706e-03, 0.000000000e+00,03，1.60941306e-02]，阵列([2.36494105e-06，1.87719732e-02，3.73829379e-03，0.00000000e + 00，4.20724542e-02，2.28675684e-02，0.00000000e + 00，5.45002249e-06, 1.17331153e-02, 1.24364600e-02])
这里是问题和需要帮助的列表:
Here it the list of questions and need help with:
- 我得到一个列表，其中只有 2 个聚类中心作为所有模型的数组，
- 当我尝试访问管道时，KMeans 模型似乎默认为 k=2?为什么会发生这种情况?
- 最后一个循环应该访问 pipelineModel 和第 0 阶段并运行 clusterCenter() 方法?这是正确的方法吗?
- 为什么会出现数据未缓存的错误?
- 这违背了使用管道并行运行 KMeans 模型和模型选择的目的，但是我尝试了以下代码:
#computeError def computeCost(model, rdd):` """Return the K-means cost (sum of squared distances of points to their nearest center) for this model on the given data.""" cost = callMLlibFunc("computeCostKmeansModel", rdd.map(_convert_to_vector), [_convert_to_vector(c) for c in model.clusterCenters()]) return cost cost= np.zeros(len(paramMap)) for i in range(len(paramMap)): cost[i] = cost[i] + computeCost(model[i].stages[0], feature_data) print cost这会在循环结束时打印出以下内容:
This prints out the following at the end of the loop:
<代码>[ 634035.00294687 634035.00294687 634035.00294687 634035.00294687634035.00294687 634035.00294687 634035.00294687 634035.00294687634035.00294687 634035.00294687 634035.00294687 634035.00294687634035.00294687 634035.00294687 634035.00294687 634035.00294687]
- 每个模型计算的成本/误差是否相同?再次无法使用正确的参数访问管道模型.
非常感谢任何帮助/指导！谢谢！
Any help/ guidance is much appreciated! Thanks!
推荐答案
您的参数定义不正确.它应该从特定参数映射到值，而不是从任意名称映射.你得到
k等于 2，因为你传递的参数没有被利用，而且每个模型都使用完全相同的默认参数.Your param is not properly defined. It should map from the specific parameters to the values, not from arbitrary names. You get
kequal 2 because parameters you pass are not utilized and every model uses exactly the same default parameters.让我们从示例数据开始:
Lets start with example data:
import numpy as np from pyspark.mllib.linalg import Vector df = (sc.textFile("data/mllib/kmeans_data.txt") .map(lambda s: Vectors.dense(np.fromstring(s, dtype=np.float64, sep=" "))) .zipWithIndex() .toDF(["features", "id"]))和一个
Pipeline:from pyspark.ml.clustering import KMeans from pyspark.ml import Pipeline km = KMeans() pipeline = Pipeline(stages=[km])如上所述，参数映射应使用特定参数作为键.例如:
As mentioned above parameter map should use specific parameters as the keys. For example:
params = [ {km.k: 2, km.initMode: "k-means||"}, {km.k: 3, km.initMode: "k-means||"}, {km.k: 4, km.initMode: "k-means||"} ] models = pipeline.fit(df, params=params) assert [len(m.stages[0].clusterCenters()) for m in models] == [2, 3, 4]注意事项:
- 正确的
initMode用于 K-means||是k-means||不是kmeans||. - 在流水线中使用参数映射并不意味着模型是并行训练的.Spark 在数据而非参数上并行化训练过程.这只不过是一种方便的方法.
- 您收到有关未缓存数据的警告，因为 K-Means 的实际输入不是
DataFrame而是转换后的 RDD.
- correct
initModefor K-means|| isk-means||notkmeans||. - using parameter map in a Pipeline doesn't mean that model are trained in parallel. Spark parallelizes training process over data not over params. It is nothing more than a convenience method.
- you get the warning about not cached data because actual input to K-Means is not a
DataFramebut transformed RDD.
这篇关于如何将参数传递给 ML Pipeline.fit 方法?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
