Download as PDF, PPTX
![tweetDF = spark.read.json('wasb:///libya-sentences/*/*.json') tokenizer = Tokenizer(inputCol="Sentence", outputCol="Words") counter = CountVectorizer(inputCol="Words", outputCol="features", vocabSize=10000, minDF=2.) lda = LDA(k=10, maxIter=10) pipeline = Pipeline(stages=[tokenizer, counter, lda]) model = pipeline.fit(tweetDF) topics = model.describeTopics(3) topics.show(truncate=False) ldaScored = model.transform(tweetDF)](https://image.slidesharecdn.com/ai04-170602095343/75/AI04-Scaling-Machine-Learning-to-Big-Data-Using-SparkML-and-SparkR-34-2048.jpg)
![tweetDF = spark.read.json('wasb:///libya-sentences/*/*.json') tweetDF = tweetDF[tweetDF.Language == 'en'] tokenizer = Tokenizer(inputCol="Sentence", outputCol="Words") enSW = StopWordsRemover.loadDefaultStopWords('english') + ['rt', '-', '&', ''] swr = StopWordsRemover(inputCol="Words", outputCol="Filtered", stopWords=enSW) tokenized = tokenizer.transform(tweetDF) filtered = swr.transform(tokenized) counter = CountVectorizer(inputCol="Filtered", outputCol=“rawFeatures", vocabSize=10000, minDF=2.) countModel = counter.fit(filtered) counted = countModel.transform(filtered)](https://image.slidesharecdn.com/ai04-170602095343/75/AI04-Scaling-Machine-Learning-to-Big-Data-Using-SparkML-and-SparkR-48-2048.jpg)
![idf = IDF(inputCol="rawFeatures", outputCol="features") idfModel = idf.fit(counted) idfScaled = idfModel.transform(counted) gbt = GBTClassifier(labelCol="label", featuresCol="features", maxIter=10) # 80/20 train/test split train, test = labeled.randomSplit([0.8, 0.2], seed=1337) model = gbt.fit(train)](https://image.slidesharecdn.com/ai04-170602095343/75/AI04-Scaling-Machine-Learning-to-Big-Data-Using-SparkML-and-SparkR-50-2048.jpg)
![%%local %matplotlib inline from sklearn.metrics import roc_curve,auc prob = predResults['prediction'] fpr, tpr, thresholds = roc_curve(predResults['label'], prob, pos_label=1); roc_auc = auc(fpr, tpr) plt.figure(figsize=(5,5)); plt.plot(fpr, tpr, label='ROC curve (area = %0.2f)' % roc_auc) plt.plot([0, 1], [0, 1], 'k--'); plt.xlim([0.0, 1.0]); plt.ylim([0.0, 1.05]) plt.xlabel('False Positive Rate'); plt.ylabel('True Positive Rate') plt.title('ROC Curve'); plt.legend(loc="lower right") plt.show()](https://image.slidesharecdn.com/ai04-170602095343/75/AI04-Scaling-Machine-Learning-to-Big-Data-Using-SparkML-and-SparkR-52-2048.jpg)
![[AI04] Scaling Machine Learning to Big Data Using SparkML and SparkR](https://image.slidesharecdn.com/ai04-170602095343/75/AI04-Scaling-Machine-Learning-to-Big-Data-Using-SparkML-and-SparkR-57-2048.jpg)
[AI04] Scaling Machine Learning to Big Data Using SparkML and SparkR
- 5. A unified, opensource, parallel data processing framework for big data analytics Spark core engine Spark SQL Interactive queries Spark Streaming Stream processing Spark ML Machine learning GraphX Graph computation Yarn Mesos Standalone scheduler
- 6.
- 8. Primary resource managers: Hadoop 1.0+or Hadoop YARN HadoopSpark Alternative resource managers: Mesos or the Spark resource manager
- 10. 102.5 100 72 23 2100 206 50400 6592 2013 Record(Hadoop) Spark 100 TB Data Size (TB) Time (Min) Nodes Cores tinyurl.com/spark-sort Logistic regression 140 120 100 80 40 20 0 60 Hadoop Spark 0.9 Logistic regression on a 100-node cluster with 100 GB of data. Spark is the 2014 Sort Benchmark winner. 3x faster than 2013 winner (Hadoop).
- 11. Reads from HDFS Writes to HDFS Readsfrom HDFS Writes to HDFS Step 1 Step 2 Step 1 Reads and writes from HDFS
- 13. ReadReadRead Cluster manager HDFS Worker nodeWorkernode Worker node Worker node Driver program SparkContext
- 15. Machine learning Real-timestream processing Developer productivity Interactive analyticsHigh performance batch computation
- 18. • .map() • .groupByKey() •.join() • .reduce() • .collect()
- 19.
- 33. tweetDF = spark.read.json('wasb:///libya-sentences/*/*.json') tokenizer= Tokenizer(inputCol="Sentence", outputCol="Words") counter = CountVectorizer(inputCol="Words", outputCol="features", vocabSize=10000, minDF=2.) tokenized = tokenizer.transform(tweetDF) countModel = counter.fit(tokenized) counted = countModel.transform(tokenized) lda = LDA(k=10, maxIter=10) model = lda.fit(counted) topics = model.describeTopics(3) topics.show(truncate=False)
- 34. tweetDF = spark.read.json('wasb:///libya-sentences/*/*.json') tokenizer= Tokenizer(inputCol="Sentence", outputCol="Words") counter = CountVectorizer(inputCol="Words", outputCol="features", vocabSize=10000, minDF=2.) lda = LDA(k=10, maxIter=10) pipeline = Pipeline(stages=[tokenizer, counter, lda]) model = pipeline.fit(tweetDF) topics = model.describeTopics(3) topics.show(truncate=False) ldaScored = model.transform(tweetDF)
- 37.
- 38.
- 43. mySparkCluster <- RxSpark() rxSetComputeContext(mySparkCluster) myData<- read.json('wasb:///creditfraud/*.json') # Run a logistic regression using RevoScaleR model <- rxLogit(Class ~ Amount + V1 + V2 + V3 + V4, data = myData) # Now run the same using SparkR Model2 <- spark.logit(myData, Class ~ Amount + V1 + V2 + V3 + V4, regParam = 0.3, elasticNetParam = 0.8) summary(model) Summary(model2)
- 48. tweetDF = spark.read.json('wasb:///libya-sentences/*/*.json') tweetDF= tweetDF[tweetDF.Language == 'en'] tokenizer = Tokenizer(inputCol="Sentence", outputCol="Words") enSW = StopWordsRemover.loadDefaultStopWords('english') + ['rt', '-', '&', ''] swr = StopWordsRemover(inputCol="Words", outputCol="Filtered", stopWords=enSW) tokenized = tokenizer.transform(tweetDF) filtered = swr.transform(tokenized) counter = CountVectorizer(inputCol="Filtered", outputCol=“rawFeatures", vocabSize=10000, minDF=2.) countModel = counter.fit(filtered) counted = countModel.transform(filtered)
- 49.
- 50. idf = IDF(inputCol="rawFeatures",outputCol="features") idfModel = idf.fit(counted) idfScaled = idfModel.transform(counted) gbt = GBTClassifier(labelCol="label", featuresCol="features", maxIter=10) # 80/20 train/test split train, test = labeled.randomSplit([0.8, 0.2], seed=1337) model = gbt.fit(train)
- 51. predictions = model.transform(test).select('prediction','label') metrics = BinaryClassificationMetrics(predictions.rdd) print('Area under PR = %s' % metrics.areaUnderPR) print('Area under ROC = %s' % metrics.areaUnderROC) # Set us up for plotting ROC predictions.registerTempTable('pred_and_labels') # In a new cell, use %%sql magic to pull results down to local context %%sql –q –o predictionResults select * from pred_and_labels
- 52. %%local %matplotlib inline from sklearn.metricsimport roc_curve,auc prob = predResults['prediction'] fpr, tpr, thresholds = roc_curve(predResults['label'], prob, pos_label=1); roc_auc = auc(fpr, tpr) plt.figure(figsize=(5,5)); plt.plot(fpr, tpr, label='ROC curve (area = %0.2f)' % roc_auc) plt.plot([0, 1], [0, 1], 'k--'); plt.xlim([0.0, 1.0]); plt.ylim([0.0, 1.05]) plt.xlabel('False Positive Rate'); plt.ylabel('True Positive Rate') plt.title('ROC Curve'); plt.legend(loc="lower right") plt.show()
- 56. HDInsight Spark SparkML HDInsight Rserver RevoScaleR SparkR GitHub Channel9 Microsoft Virtual Academy