1
$\begingroup$

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?

$\endgroup$
  • $\begingroup$ Are you trying to collect your data ? $\endgroup$ – LaSul Feb 6 '19 at 9:12
  • $\begingroup$ so i have csv file, which i'm importing and all, everything is happening fine until I try to fit my model in the algo from the PySpark package $\endgroup$ – Debadri Dutta Feb 6 '19 at 9:25
  • $\begingroup$ Okay thank. Could you now add sample code please ? $\endgroup$ – LaSul Feb 6 '19 at 11:20
  • $\begingroup$ hey, added can you please check and give me any idea? But why is that for say datasets having 5k-6k values, sklearn Random Forest works fine but PySpark random forest fails? Is it a way that PySpark dataframe stores the features? If yes, how can I solve this issue? $\endgroup$ – Debadri Dutta Feb 7 '19 at 5:04
  • $\begingroup$ Okay, I don't see any issue here, can you tell me how you define sqlContext ? You might need to increase driver & executor memory size $\endgroup$ – LaSul Feb 7 '19 at 10:02
0
$\begingroup$

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.

| improve this answer | |
$\endgroup$
-1
$\begingroup$

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.

| improve this answer | |
$\endgroup$
  • $\begingroup$ Pandas worked. Spark failed $\endgroup$ – Has QUIT--Anony-Mousse Mar 14 '19 at 23:05
  • $\begingroup$ Not true. If pandas tries to fit anything in memory which doesn't fit it, there would be a memory error. So, you can either assign more resources to let the code use more memory/you'll have to loop, like @Debadri Dutta is doing. When you assign more resources, you're limiting other resources on your computer from using that memory. Assign too much, and it would hang up and fail to do anything else, really. So, if you know that the data is going to increase, you should look into the options of expanding into Pyspark. $\endgroup$ – user2816215 Mar 15 '19 at 19:05
  • $\begingroup$ Read his question again. Spark already failed after some 3-4k rows, pandas managed 20k (both of which I would consider tiny, but since it is tfidf it likely has a lot of columns. And it seems the only thing that Spark is actually able to beat reliably is old MapReduce... Too much hype. $\endgroup$ – Has QUIT--Anony-Mousse Mar 15 '19 at 19:13

Your Answer

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

Not the answer you're looking for? Browse other questions tagged or ask your own question.