如何将参数传递给ML管道.fit方法？

#!/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()

#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

#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

1条回答

网友

1楼 · 发布于 2024-10-04 05:20:00

参数定义不正确。它应该从特定参数映射到值，而不是从任意名称映射。您得到k等于2，因为您传递的参数没有被利用，而且每个模型使用完全相同的默认参数。在

让我们从示例数据开始：

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：

^{pr2}$

如上所述，参数映射应该使用特定的参数作为键。例如：

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]

注意事项：

K-均值的正确initMode是k-means||不是{}。在
在流水线中使用参数映射并不意味着模型是并行训练的。Spark将训练过程与数据并行，而不是参数。这只不过是一种方便的方法。在
您会收到关于未缓存数据的警告，因为K-Means的实际输入不是DataFrame，而是经过转换的RDD。在

相关问题更多 >

编程相关推荐

热门问题

热门文章