I'm experimenting with the RandomForestRegressor. The main dataframe created by Aggregations on Windows over Event-Time. Let's say below are train and test sets in the form of the dataframes including timestamp column (type struct):
#trainset dataframe
+------------------------------------------+----+.....+----+
|window_frame_24_Hours |A | ... |Z |
+------------------------------------------+---------------+
|{2021-11-28 00:00:00, 2021-11-29 00:00:00}|316 | ... |666 |
|{2021-11-27 00:00:00, 2021-11-28 00:00:00}|324 | ... |526 |
|{2021-11-26 00:00:00, 2021-11-27 00:00:00}|261 | ... |414 |
|{2021-11-25 00:00:00, 2021-11-26 00:00:00}|268 | ... |632 |
|{2021-11-24 00:00:00, 2021-11-25 00:00:00}|284 | ... |578 |
|{2021-11-23 00:00:00, 2021-11-24 00:00:00}|232 | ... |226 |
... .... ...
|{2021-11-02 00:00:00, 2021-11-03 00:00:00}|94 | ... |100 |
|{2021-11-01 00:00:00, 2021-11-02 00:00:00}|106 | ... |666 |
|{2021-10-31 00:00:00, 2021-11-01 00:00:00}|130 | ... |108 |
|{2021-10-30 00:00:00, 2021-10-31 00:00:00}|112 | ... |35 |
+------------------------------------------+----+.....+----+
#testset dataframe
+------------------------------------------+----+.....+----+
|window_frame_24_Hours |A | ... |Z |
+------------------------------------------+---------------+
|{2021-11-28 00:00:00, 2021-11-29 00:00:00}|?? | ... |?? | <-- predict
+------------------------------------------+----+.....+----+
following is my RF model implementation:
#Dependencies
from pyspark.ml import Pipeline
from pyspark.ml.feature import StringIndexer
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.feature import MinMaxScaler
from pyspark.ml.regression import RandomForestRegressor, GBTRegressor
from pyspark.ml.tuning import CrossValidator, ParamGridBuilder
value = X_train.columns[1:]
assembler = VectorAssembler(inputCols=value, outputCol='features')
scaler = MinMaxScaler(inputCol='features', outputCol='scaled_features')
# Create a RandomForest model.
rf = RandomForestRegressor(featuresCol="scaled_features", labelCol="label") #labelCol=value
# Chain model, assembler and scaler into a Pipeline.
pipeline = Pipeline(stages=[assembler, scaler, rf])
# Train model on training data.
rf_model = pipeline.fit(X_train)
# Make predictions.
predictions = rf_model.transform(X_test)
# Select example rows to display.
predictions = predictions.select("window_frame_24_Hours", "prediction", "features", "scaled_features").sort('window_frame_24_Hours.start')
##predictions.show(45, truncate = False)
I want to predict columns A - Z over certain time so I exclude timestamp column by X_train.columns[1:] and pass target columns via a VectorAssembler to assembler. Then:
when I use labelCol=value in rf = RandomForestRegressor(featuresCol="scaled_features", labelCol=value) I face:
TypeError: Invalid param value given for param "labelCol". Could not convert <class 'list'> to string type
when I use labelCol="label" in rf = RandomForestRegressor(featuresCol="scaled_features", labelCol=value) I face:
AnalysisException: Cannot resolve column name "A" among (window_frame_24_Hours, A, ..., Z,
the question:
What is the best practice to fit time-series based dataframe to predict multiple columns. (multi-regression problem)
Should I use for-loop and train for each column based on this disappointing post for spark? if so, how?
I tried to check other time-series regression examples 1, 2 & 3 as well as this video but their problem was not multi-regressions (predict lots of columns) nor it's not clear how they handle it. I know in in python there is solution for Multioutput regression.