When running a logistic regression on a large dataset with repartition, how do we stitch the results of the each partition results back to a dataframe to analyze.

from pyspark.ml.feature import StringIndexer, VectorAssembler, OneHotEncoder, StandardScaler
from pyspark.ml import Pipeline
from pyspark.ml.classification import LogisticRegression

model_indexing = StringIndexer(inputCol= 'model', outputCol= 'model_index')

one_hot_encoding_model = OneHotEncoder(inputCol="model_index", outputCol="model_encoded")

label_stringIndexer = StringIndexer(inputCol = 'is_usage_higher', outputCol = "label").setHandleInvalid("skip")

assembler_inputs  = model_columns +  ['model_encoded']

assembler = VectorAssembler(inputCols = assembler_inputs,
                            outputCol = "features")

scaler = StandardScaler(inputCol = "features",  outputCol = "scaledFeatures")

Stages = [model_indexing, one_hot_encoding_model, label_stringIndexer, assembler, scaler]

lr_model = LogisticRegression(featuresCol='scaledFeatures',labelCol='label')

def model_parallelly(partition_df, model_pipe):
    train, test = partition_df.randomSplit([0.8,0.2])
    modelPipeline = Pipeline().setStages(Stages + [model_pipe])
    run_model = modelPipeline.fit(train)
    predictions = run_model.transform(test)
    return predictions

model_partition_df = df.repartition('hour')

lr_model_result =  model_partition_df.rdd.mapPartitions(lambda part: model_parallelly(part, lr_model))

What are different ways to run a ML model in parallel on a large dataset


Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.