【Apache Spark】Spark MLLib - Predict Store Sales with ML Pipelines

是 否 +2 论坛币

k人 参与回答

经管之家送您一份

应届毕业生专属福利!

求职就业群

赵安豆老师微信：zhaoandou666

经管之家联合CDA

送您一个全额奖学金名额~ !

立即领取

感谢您参与论坛问题回答

经管之家送您两个论坛币！

+2 论坛币

本帖隐藏的内容

Spark MLLib - Predict Store Sales with ML Pipelines.pdf (474.22 KB)

2017-4-30 09:38:00 上传

1. Overview
   
2. Recently I had to work on a Machine Learning problem for class and found a good opportunity for a Spark Tutorial. Using store sales from Rossmann found on kaggle, we are going to set up a machine learning pipeline to cover everything from the preprocessing all the way to making and saving predictions. In a sense, this is a "full-stack" machine learning project that's probably fairly similar to something we might do in the real world. Spark's ML Pipelines API is going to make it very easy for us to do this. You can follow along with the code on my github or below.

复制代码

扫码加我 拉你入群

请注明：姓名-公司-职位

以便审核进群资格，未注明则拒绝

分享0 收藏0 回帖

关键词：Apache Spark Pipelines pipeline predict apache

1. Linear Regression
   
2. The pipeline we've set up is pretty self explanatory. Scaladef preppedLRPipeline():TrainValidationSplit = {
   
3.   val lr = new LinearRegression()
   
4. 
   
5.   val paramGrid = new ParamGridBuilder()
   
6.     .addGrid(lr.regParam, Array(0.1, 0.01))
   
7.     .addGrid(lr.fitIntercept)
   
8.     .addGrid(lr.elasticNetParam, Array(0.0, 0.25, 0.5, 0.75, 1.0))
   
9.     .build()
   
10. 
    
11.   val pipeline = new Pipeline()
    
12.     .setStages(Array(stateHolidayIndexer, schoolHolidayIndexer,
    
13.       stateHolidayEncoder, schoolHolidayEncoder, storeEncoder,
    
14.       dayOfWeekEncoder, dayOfMonthEncoder,
    
15.       assembler, lr))
    
16. 
    
17.   val tvs = new TrainValidationSplit()
    
18.     .setEstimator(pipeline)
    
19.     .setEvaluator(new RegressionEvaluator)
    
20.     .setEstimatorParamMaps(paramGrid)
    
21.     .setTrainRatio(0.75)
    
22.   tvs
    
23. }

复制代码

1. def loadTrainingData(sqlContext:HiveContext):DataFrame = {
   
2.   val trainRaw = sqlContext
   
3.     .read.format("com.databricks.spark.csv")
   
4.     .option("header", "true")
   
5.     .load("../mlproject/rossman/train.csv")
   
6.     .repartition(6)
   
7.   trainRaw.registerTempTable("raw_training_data")
   
8. 
   
9.   sqlContext.sql("""SELECT
   
10.     double(Sales) label, double(Store) Store, int(Open) Open, double(DayOfWeek) DayOfWeek,
    
11.     StateHoliday, SchoolHoliday, (double(regexp_extract(Date, '\\d+-\\d+-(\\d+)', 1))) DayOfMonth
    
12.     FROM raw_training_data
    
13.   """).na.drop()
    
14. }
    
15. 
    
16. def loadKaggleTestData(sqlContext:HiveContext) = {
    
17.   val testRaw = sqlContext
    
18.     .read.format("com.databricks.spark.csv")
    
19.     .option("header", "true")
    
20.     .load("../mlproject/rossman/test.csv")
    
21.     .repartition(6)
    
22.   testRaw.registerTempTable("raw_test_data")
    
23. 
    
24.   val testData = sqlContext.sql("""SELECT
    
25.     Id, double(Store) Store, int(Open) Open, double(DayOfWeek) DayOfWeek, StateHoliday,
    
26.     SchoolHoliday, (double(regexp_extract(Date, '\\d+-\\d+-(\\d+)', 1))) DayOfMonth
    
27.     FROM raw_test_data
    
28.     WHERE !(ISNULL(Id) OR ISNULL(Store) OR ISNULL(Open) OR ISNULL(DayOfWeek)
    
29.       OR ISNULL(StateHoliday) OR ISNULL(SchoolHoliday))
    
30.   """).na.drop() // weird things happen if you don't filter out the null values manually
    
31. 
    
32.   Array(testRaw, testData) // got to hold onto testRaw so we can make sure
    
33.   // to have all the prediction IDs to submit to kaggle
    
34. }

复制代码

1. def savePredictions(predictions:DataFrame, testRaw:DataFrame) = {
   
2.   val tdOut = testRaw
   
3.     .select("Id")
   
4.     .distinct()
   
5.     .join(predictions, testRaw("Id") === predictions("PredId"), "outer")
   
6.     .select("Id", "Sales")
   
7.     .na.fill(0:Double) // some of our inputs were null so we have to
   
8.                        // fill these with something
   
9.   tdOut
   
10.     .coalesce(1)
    
11.     .write.format("com.databricks.spark.csv")
    
12.     .option("header", "true")
    
13.     .save("linear_regression_predictions.csv")
    
14. }

复制代码

1. def fitModel(tvs:TrainValidationSplit, data:DataFrame) = {
   
2.   val Array(training, test) = data.randomSplit(Array(0.8, 0.2), seed = 12345)
   
3.   logger.info("Fitting data")
   
4.   val model = tvs.fit(training)
   
5.   logger.info("Now performing test on hold out set")
   
6.   val holdout = model.transform(test).select("prediction","label")
   
7. 
   
8.   // have to do a type conversion for RegressionMetrics
   
9.   val rm = new RegressionMetrics(holdout.rdd.map(x =>
   
10.     (x(0).asInstanceOf[Double], x(1).asInstanceOf[Double])))
    
11. 
    
12.   logger.info("Test Metrics")
    
13.   logger.info("Test Explained Variance:")
    
14.   logger.info(rm.explainedVariance)
    
15.   logger.info("Test R^2 Coef:")
    
16.   logger.info(rm.r2)
    
17.   logger.info("Test MSE:")
    
18.   logger.info(rm.meanSquaredError)
    
19.   logger.info("Test RMSE:")
    
20.   logger.info(rm.rootMeanSquaredError)
    
21. 
    
22.   model
    
23. }

复制代码

1. val data = loadTrainingData(sqlContext)
   
2. val Array(testRaw, testData) = loadKaggleTestData(sqlContext)
   
3. 
   
4. // The linear Regression Pipeline
   
5. val linearTvs = preppedLRPipeline()
   
6. logger.info("evaluating linear regression")
   
7. val lrModel = fitModel(linearTvs, data)
   
8. logger.info("Generating kaggle predictions")
   
9. val lrOut = lrModel.transform(testData)
   
10.   .withColumnRenamed("prediction","Sales")
    
11.   .withColumnRenamed("Id","PredId")
    
12.   .select("PredId", "Sales")
    
13. savePredictions(lrOut, testRaw)

复制代码

多谢分享！

非常感谢分享！

谢谢楼主分享！