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?
collect
your data ? $\endgroup$