PySpark培訓隨機森林管道

Question

我正在嘗試創建一個管道，需要我的DataFrame的航班延誤信息並在其上運行隨機森林。我對MLLib相當陌生，無法弄清楚我的代碼在下面出錯。

我的數據幀從一個木文件中讀取這種格式：

Table before Encoding 
+-----+-----+---+---+----+--------+-------+------+----+-----+-------+ 
|Delay|Month|Day|Dow|Hour|Distance|Carrier|Origin|Dest|HDays|Delayed| 
+-----+-----+---+---+----+--------+-------+------+----+-----+-------+ 
| -8| 8| 4| 2| 11|  224|  OO| GEG| SEA| 31|  0| 
| -12| 8| 5| 3| 11|  224|  OO| GEG| SEA| 32|  0| 
| -9| 8| 6| 4| 11|  224|  OO| GEG| SEA| 32|  0| 
+-----+-----+---+---+----+--------+-------+------+----+-----+-------+ 
only showing top 3 rows 

我然後進行OneHotEncode的類別列，並且所有的功能合併到一個Features柱（Delayed是我想要預測）。下面是該代碼：

import os 
from pyspark.sql import SparkSession 
from pyspark.ml import Pipeline 
from pyspark.ml.feature import OneHotEncoder, StringIndexer, VectorAssembler 
from pyspark.ml.classification import LogisticRegression, RandomForestClassifier 

spark = SparkSession.builder \ 
    .master('local[3]') \ 
    .appName('Flight Delay') \ 
    .getOrCreate() 

# read in the pre-processed DataFrame from the parquet file 
base_dir = '/home/william/Projects/flight-delay/data/parquet' 
flights_df = spark.read.parquet(os.path.join(base_dir, 'flights.parquet')) 

print('Table before Encoding') 
flights_df.show(3) 

# categorical columns that will be OneHotEncoded 
cat_cols = ['Month', 'Day', 'Dow', 'Hour', 'Carrier', 'Dest'] 

# numeric columns that will be a part of features used for prediction 
non_cat_cols = ['Delay', 'Distance', 'HDays'] 

# NOTE: StringIndexer does not have multiple col support yet (PR #9183) 
# Create StringIndexer for each categorical feature 
cat_indexers = [ StringIndexer(inputCol=col, outputCol=col+'_Index') 
       for col in cat_cols ] 

# OneHotEncode each categorical feature after being StringIndexed 
encoders = [ OneHotEncoder(dropLast=False, inputCol=indexer.getOutputCol(), 
          outputCol=indexer.getOutputCol()+'_Encoded') 
      for indexer in cat_indexers ] 

# Assemble all feature columns (numeric + categorical) into `features` col 
assembler = VectorAssembler(inputCols=[encoder.getOutputCol() 
             for encoder in encoders] + non_cat_cols, 
          outputCol='Features') 

# Train a random forest model 
rf = RandomForestClassifier(labelCol='Delayed',featuresCol='Features', numTrees=10) 

# Chain indexers, encoders, and forest into one pipeline 
pipeline = Pipeline(stages=[ *cat_indexers, *encoders, assembler, rf ]) 

# split the data into training and testing splits (70/30 rn) 
(trainingData, testData) = flights_df.randomSplit([0.7, 0.3]) 

# Train the model -- which also runs indexers and coders 
model = pipeline.fit(trainingData) 

# use model to make predictions 
precitions = model.trainsform(testData) 

predictions.show(10) 

當我運行此我得到一個 Py4JJavaError: An error occurred while calling o46.fit. : java.lang.ClassCastException: java.lang.Integer cannot be cast to java.lang.Double

我非常感謝任何幫助！

Answer 1

正如所解釋的in the comments，標籤應double，所以你要投：

flights_df = spark.read.parquet(os.path.join(base_dir, 'flights.parquet')) \ 
    .withColumn("Delayed", col("Delayed").cast("double"))