PySpark v Pandas Dataframe Memory Issue

Question

Suppose I have a csv file with 20k rows, which I import into Pandas dataframe. I then run models like Random Forest or Logistic Regression from sklearn package and it runs fine. However, when I import into PySpark dataframe format and run the same models (Random Forest or Logistic Regression) from PySpark packages, I get a memory error and I have to reduce the size of the csv down to say 3-4k rows. Why does this happen? Is this a conceptual problem or am I coding it wrong somewhere?

For Pandas dataframe, my sample code is something like this:

df=pd.read_csv("xx.csv")

features=TfIdf().fit(df['text'])
....
RandomForest.fit(features,labels)

And for PySpark, I'm first reading the file like this:

data = sqlContext.read.format('com.databricks.spark.csv').options(header='true', inferschema='true')\
.load('xx.csv')
data.show()
from pyspark.ml.feature import RegexTokenizer, StopWordsRemover, CountVectorizer
from pyspark.ml.classification import LogisticRegression
# regular expression tokenizer
regexTokenizer = RegexTokenizer(inputCol="converted_text", outputCol="words", pattern="\\W")
# stop words
add_stopwords = ["http","https","amp","rt","t","c","the"] 
stopwordsRemover = StopWordsRemover(inputCol="words", outputCol="filtered").setStopWords(add_stopwords)
# bag of words count
countVectors = CountVectorizer(inputCol="filtered", outputCol="features", vocabSize=10000, minDF=5)
from pyspark.ml import Pipeline
from pyspark.ml.feature import OneHotEncoder, StringIndexer, VectorAssembler
label_stringIdx = StringIndexer(inputCol = "Complaint-Status", outputCol = "label")
pipeline = Pipeline(stages=[regexTokenizer, stopwordsRemover, countVectors, label_stringIdx])
# Fit the pipeline to training documents.
pipelineFit = pipeline.fit(data)
dataset = pipelineFit.transform(data)
dataset.show(5)
(trainingData, testData) = dataset.randomSplit([0.7, 0.3], seed = 100)
print("Training Dataset Count: " + str(trainingData.count()))
print("Test Dataset Count: " + str(testData.count()))
from pyspark.ml.classification import RandomForestClassifier
rf = RandomForestClassifier(labelCol="label", \
                            featuresCol="features", \
                            numTrees = 100, \
                            maxDepth = 4, \
                            maxBins = 32)
# Train model with Training Data
rfModel = rf.fit(trainingData)
predictions = rfModel.transform(testData)
predictions.filter(predictions['prediction'] == 0) \
    .select("converted_text","Complaint-Status","probability","label","prediction") \
    .orderBy("probability", ascending=False) \
    .show(n = 10, truncate = 30)

I was trying for lightgbm, only changing the .fit() part:

from mmlspark import LightGBMClassifier
lgb = LightGBMClassifier(learningRate=0.3,
                           numIterations=100,
                           numLeaves=31)
lgb_model=lgb.fit(trainingData)
predictions = lgb_model.transform(testData)
predictions.filter(predictions['prediction'] == 0) \
    .select("converted_text","Complaint-Status","probability","label","prediction") \
    .orderBy("probability", ascending=False) \
    .show(n = 10, truncate = 30)
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
evaluator = MulticlassClassificationEvaluator(predictionCol="prediction")
evaluator.evaluate(predictions)

And the dataset has hardly 5k rows inside the csv files. Why is it happening? How can I solve it?

Answer 1

While I can't tell you why Spark is so slow (it does come with overheads, and it only makes sense to use Spark when you have 20+ nodes in a big cluster and data that does not fit into RAM of a single PC - unless you use distributed processing, the overheads will cause such problems. For example, your program first has to copy all the data into Spark, so it will need at least twice as much memory. Probably even three copies: your original data, the pyspark copy, and then the Spark copy in the JVM. In the worst case, the data is transformed into a dense format when doing so, at which point you may easily waste 100x as much memory because of storing all the zeros).

Use an appropriate - smaller - vocabulary.

There is no use in including every single word, as most of them will never score well in the decision trees anyway!

It's safe to assume that you can omit both very frequent (stop-) words, as well as rare words (using them would be overfitting anyway!). So use min_df=10 and max_df=1000 or so.

Answer 2

Pandas dataframes can be rather fickle. I've observed code running fine until one line somewhere tries to load more data in memory than it can handle and it all breaks apart, landing a memory error. Even if the rows are limited, the number of columns and the content of each cell also matters. If it's all long strings, the data can be more than pandas can handle. Now, if you train using fit on all of that data, it might not fit in the memory at once. Look here for one previous answer.

Pyspark, on the other hand, has been optimized for handling 'big data'. An rdd contains many partitions, which may be distributed and it can spill files to disk. Here, you can read more on it.