cache-for-correctness.ipynb

{"nbformat_minor": 0, "cells": [{"source": "# Readme\n\nReading the blog [Using Spark's cache for correctness, not just performance](http://www.spark.tc/using-sparks-cache-for-correctness-not-just-performance/) and trying the code below.  \nYou can try this notebook on IBM **Bluemix** using [IBM Starter for Apache Spark](https://console.ng.bluemix.net/catalog/apache-spark-starter/).  ", "cell_type": "markdown", "metadata": {}}, {"execution_count": 7, "cell_type": "code", "source": "import org.apache.spark.SparkConf\nimport org.apache.spark.SparkContext\nimport scala.util.Random\n\n// start with a sequence of 10,000 zeros\nval zeros = Seq.fill(10000)(0)\n\n// create a RDD from the sequence, and replace all zeros with random values\nval randomRDD = sc.parallelize(zeros).map(x=>Random.nextInt())\n\n// filter out all non-positive values, roughly half the set\nval filteredRDD = randomRDD.filter(x=>x>0)\n\n// count the number of elements that remain, twice\nval count1 = filteredRDD.count()\nval count2 = filteredRDD.count()\n\nprintln(s\"Surprise! count1: ${count1}, count2: ${count2}\")", "outputs": [{"output_type": "stream", "name": "stdout", "text": "Surprise! count1: 4863, count2: 4953\n"}], "metadata": {"collapsed": false}}, {"source": "## Would you expect that `count1` and `count2` are equals, right?  \n\nWell, that rarely happens because the `map` and `filter` are **transformations** and `count` is an **action**, which\nmeans that any time `count` is called, the entire pipeline of transformations is executed again. So, the `randomRDD`\nchanges and so does `filteredRDD`.  \n\n## So, what?  \n\nWhen you are in these situations, you should `cache (or persist)` RDD, so [Using Spark's cache for correctness, not just performance](http://www.spark.tc/using-sparks-cache-for-correctness-not-just-performance/).  ", "cell_type": "markdown", "metadata": {}}, {"execution_count": 8, "cell_type": "code", "source": "import org.apache.spark.storage.StorageLevel\nimport org.apache.spark.SparkConf\nimport org.apache.spark.SparkContext\nimport scala.util.Random\n\n// start with a sequence of 10,000 zeros\nval zeros = Seq.fill(10000)(0)\n\n// create a RDD from the sequence, and replace all zeros with random values\nval randomRDD = sc.parallelize(zeros).map(x=>Random.nextInt())\n\n// filter out all non-positive values, roughly half the set\nval filteredRDD = randomRDD.filter(x=>x>0).persist(StorageLevel.MEMORY_AND_DISK)\n\n// count the number of elements that remain, twice\nval count1 = filteredRDD.count()\nval count2 = filteredRDD.count()\n\nprintln(s\"No alarms and no surprises! count1: ${count1}, count2: ${count2}\")", "outputs": [{"output_type": "stream", "name": "stdout", "text": "No alarms and no surprises! count1: 5035, count2: 5035\n"}], "metadata": {"collapsed": false}}, {"execution_count": null, "cell_type": "code", "source": "", "outputs": [], "metadata": {"collapsed": true}}], "nbformat": 4, "metadata": {"kernelspec": {"display_name": "Scala 2.10", "name": "spark", "language": "scala"}, "language_info": {"name": "scala"}, "name": "Cache for correctness.ipynb"}}