Streaming_percentile.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Warning: FastBinaryTree not available, using Python version BinaryTree.\n",
      "Warning: FastAVLTree not available, using Python version AVLTree.\n",
      "Warning: FastRBTree not available, using Python version RBTree.\n"
     ]
    }
   ],
   "source": [
    "import json\n",
    "from time import time\n",
    "from operator import add\n",
    "import argparse\n",
    "\n",
    "from tdigest import TDigest\n",
    "from kafka import KeyedProducer, RoundRobinPartitioner, KafkaClient\n",
    "\n",
    "from pyspark.streaming.kafka import KafkaUtils\n",
    "from pyspark.streaming import StreamingContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "percentile_broadcast = None\n",
    "\n",
    "\n",
    "def load_msg(msg):\n",
    "    message = json.loads(msg[1])\n",
    "    return message['user_id'], scores_b.value[message['activity']]\n",
    "\n",
    "\n",
    "def update_scorecount(new_scores, score_sum):\n",
    "    if not score_sum:\n",
    "        score_sum = 0\n",
    "    return sum(new_scores) + score_sum\n",
    "\n",
    "\n",
    "def digest_partitions(values):\n",
    "    digest = TDigest()\n",
    "    digest.batch_update(values)\n",
    "    return [digest]\n",
    "\n",
    "\n",
    "def publish_popular_users(popular_rdd):\n",
    "    key = 'popular_{}'.format(int(time()))\n",
    "    message_key = popular_rdd.context.broadcast(key)\n",
    "\n",
    "    def publish_partition(partition):\n",
    "        kafka = KafkaClient('kafka:9092')\n",
    "        producer = KeyedProducer(kafka, partitioner=RoundRobinPartitioner,\n",
    "                                 async=True, batch_send=True)\n",
    "        producer.send_messages('popular_users', message_key.value,\n",
    "                               *[json.dumps(user) for user in partition])\n",
    "    popular_rdd.foreachPartition(publish_partition)\n",
    "\n",
    "\n",
    "def compute_percentile(rdd):\n",
    "    global percentile_broadcast\n",
    "    percentile_limit = rdd.map(lambda row: row[1]).mapPartitions(\n",
    "        digest_partitions).reduce(add).percentile(0.95)\n",
    "    percentile_broadcast = rdd.context.broadcast(\n",
    "        percentile_limit)\n",
    "\n",
    "\n",
    "def filter_most_popular(rdd):\n",
    "    global percentile_broadcast\n",
    "    if percentile_broadcast:\n",
    "        return rdd.filter(lambda row: row[1] > percentile_broadcast.value)\n",
    "    return rdd.context.parallelize([])\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "ssc = StreamingContext(sc, 1)\n",
    "ssc.checkpoint('checkpoint')\n",
    "kvs = KafkaUtils.createDirectStream(ssc, ['messages'], {\"metadata.broker.list\": \"kafka:9092\"})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Traceback (most recent call last):\n",
      "  File \"/usr/local/spark/python/pyspark/streaming/util.py\", line 62, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/usr/local/spark/python/pyspark/streaming/dstream.py\", line 159, in <lambda>\n",
      "    func = lambda t, rdd: old_func(rdd)\n",
      "  File \"<ipython-input-2-21c82552202c>\", line 31, in publish_popular_users\n",
      "    popular_rdd.foreachPartition(publish_partition)\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 766, in foreachPartition\n",
      "    self.mapPartitions(func).count()  # Force evaluation\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 1006, in count\n",
      "    return self.mapPartitions(lambda i: [sum(1 for _ in i)]).sum()\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 997, in sum\n",
      "    return self.mapPartitions(lambda x: [sum(x)]).fold(0, operator.add)\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 871, in fold\n",
      "    vals = self.mapPartitions(func).collect()\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 773, in collect\n",
      "    port = self.ctx._jvm.PythonRDD.collectAndServe(self._jrdd.rdd())\n",
      "  File \"/usr/local/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 538, in __call__\n",
      "    self.target_id, self.name)\n",
      "  File \"/usr/local/spark/python/pyspark/sql/utils.py\", line 36, in deco\n",
      "    return f(*a, **kw)\n",
      "  File \"/usr/local/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py\", line 300, in get_return_value\n",
      "    format(target_id, '.', name), value)\n",
      "Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.\n",
      ": org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 2.0 failed 1 times, most recent failure: Lost task 0.0 in stage 2.0 (TID 9, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/worker.py\", line 111, in main\n",
      "    process()\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/worker.py\", line 106, in process\n",
      "    serializer.dump_stream(func(split_index, iterator), outfile)\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 2355, in pipeline_func\n",
      "    return func(split, prev_func(split, iterator))\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 2355, in pipeline_func\n",
      "    return func(split, prev_func(split, iterator))\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 2355, in pipeline_func\n",
      "    return func(split, prev_func(split, iterator))\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 317, in func\n",
      "    return f(iterator)\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 761, in func\n",
      "    r = f(it)\n",
      "  File \"<ipython-input-2-21c82552202c>\", line 30, in publish_partition\n",
      "  File \"/usr/local/lib/python2.7/dist-packages/kafka/producer/keyed.py\", line 42, in send_messages\n",
      "    partition = self._next_partition(topic, key)\n",
      "  File \"/usr/local/lib/python2.7/dist-packages/kafka/producer/keyed.py\", line 33, in _next_partition\n",
      "    self.client.load_metadata_for_topics(topic)\n",
      "  File \"/usr/local/lib/python2.7/dist-packages/kafka/client.py\", line 378, in load_metadata_for_topics\n",
      "    kafka.common.check_error(topic_metadata)\n",
      "  File \"/usr/local/lib/python2.7/dist-packages/kafka/common.py\", line 233, in check_error\n",
      "    raise error_class(response)\n",
      "LeaderNotAvailableError: TopicMetadata(topic='popular_users', error=5, partitions=[])\n",
      "\n",
      "\tat org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:166)\n",
      "\tat org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:207)\n",
      "\tat org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:125)\n",
      "\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:70)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)\n",
      "\tat org.apache.spark.scheduler.Task.run(Task.scala:88)\n",
      "\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)\n",
      "\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)\n",
      "\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)\n",
      "\tat java.lang.Thread.run(Thread.java:744)\n",
      "\n",
      "Driver stacktrace:\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1283)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1271)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1270)\n",
      "\tat scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)\n",
      "\tat scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1270)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:697)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:697)\n",
      "\tat scala.Option.foreach(Option.scala:236)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:697)\n",
      "\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1496)\n",
      "\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1458)\n",
      "\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1447)\n",
      "\tat org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:567)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1822)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1835)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1848)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1919)\n",
      "\tat org.apache.spark.rdd.RDD$$anonfun$collect$1.apply(RDD.scala:905)\n",
      "\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:147)\n",
      "\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:108)\n",
      "\tat org.apache.spark.rdd.RDD.withScope(RDD.scala:306)\n",
      "\tat org.apache.spark.rdd.RDD.collect(RDD.scala:904)\n",
      "\tat org.apache.spark.api.python.PythonRDD$.collectAndServe(PythonRDD.scala:405)\n",
      "\tat org.apache.spark.api.python.PythonRDD.collectAndServe(PythonRDD.scala)\n",
      "\tat sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\n",
      "\tat sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)\n",
      "\tat sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\n",
      "\tat java.lang.reflect.Method.invoke(Method.java:606)\n",
      "\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231)\n",
      "\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:379)\n",
      "\tat py4j.Gateway.invoke(Gateway.java:259)\n",
      "\tat py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)\n",
      "\tat py4j.commands.CallCommand.execute(CallCommand.java:79)\n",
      "\tat py4j.GatewayConnection.run(GatewayConnection.java:207)\n",
      "\tat java.lang.Thread.run(Thread.java:744)\n",
      "Caused by: org.apache.spark.api.python.PythonException: Traceback (most recent call last):\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/worker.py\", line 111, in main\n",
      "    process()\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/worker.py\", line 106, in process\n",
      "    serializer.dump_stream(func(split_index, iterator), outfile)\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 2355, in pipeline_func\n",
      "    return func(split, prev_func(split, iterator))\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 2355, in pipeline_func\n",
      "    return func(split, prev_func(split, iterator))\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 2355, in pipeline_func\n",
      "    return func(split, prev_func(split, iterator))\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 317, in func\n",
      "    return f(iterator)\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 761, in func\n",
      "    r = f(it)\n",
      "  File \"<ipython-input-2-21c82552202c>\", line 30, in publish_partition\n",
      "  File \"/usr/local/lib/python2.7/dist-packages/kafka/producer/keyed.py\", line 42, in send_messages\n",
      "    partition = self._next_partition(topic, key)\n",
      "  File \"/usr/local/lib/python2.7/dist-packages/kafka/producer/keyed.py\", line 33, in _next_partition\n",
      "    self.client.load_metadata_for_topics(topic)\n",
      "  File \"/usr/local/lib/python2.7/dist-packages/kafka/client.py\", line 378, in load_metadata_for_topics\n",
      "    kafka.common.check_error(topic_metadata)\n",
      "  File \"/usr/local/lib/python2.7/dist-packages/kafka/common.py\", line 233, in check_error\n",
      "    raise error_class(response)\n",
      "LeaderNotAvailableError: TopicMetadata(topic='popular_users', error=5, partitions=[])\n",
      "\n",
      "\tat org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:166)\n",
      "\tat org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:207)\n",
      "\tat org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:125)\n",
      "\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:70)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)\n",
      "\tat org.apache.spark.scheduler.Task.run(Task.scala:88)\n",
      "\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)\n",
      "\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)\n",
      "\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)\n",
      "\t... 1 more\n",
      "\n",
      "Traceback (most recent call last):\n",
      "  File \"/usr/local/spark/python/pyspark/streaming/util.py\", line 62, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/usr/local/spark/python/pyspark/streaming/dstream.py\", line 159, in <lambda>\n",
      "    func = lambda t, rdd: old_func(rdd)\n",
      "  File \"<ipython-input-2-21c82552202c>\", line 37, in compute_percentile\n",
      "    digest_partitions).reduce(add).percentile(0.95)\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 799, in reduce\n",
      "    vals = self.mapPartitions(func).collect()\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 773, in collect\n",
      "    port = self.ctx._jvm.PythonRDD.collectAndServe(self._jrdd.rdd())\n",
      "  File \"/usr/local/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 538, in __call__\n",
      "    self.target_id, self.name)\n",
      "  File \"/usr/local/spark/python/pyspark/sql/utils.py\", line 36, in deco\n",
      "    return f(*a, **kw)\n",
      "  File \"/usr/local/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py\", line 300, in get_return_value\n",
      "    format(target_id, '.', name), value)\n",
      "Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.\n",
      ": org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 1841.0 failed 1 times, most recent failure: Lost task 0.0 in stage 1841.0 (TID 2526, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/worker.py\", line 98, in main\n",
      "    command = pickleSer._read_with_length(infile)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/serializers.py\", line 164, in _read_with_length\n",
      "    return self.loads(obj)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/serializers.py\", line 422, in loads\n",
      "    return pickle.loads(obj)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/broadcast.py\", line 39, in _from_id\n",
      "    raise Exception(\"Broadcast variable '%s' not loaded!\" % bid)\n",
      "Exception: (Exception(\"Broadcast variable '0' not loaded!\",), <function _from_id at 0x7f923b0da5f0>, (0L,))\n",
      "\n",
      "\tat org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:166)\n",
      "\tat org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:207)\n",
      "\tat org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:125)\n",
      "\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:70)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:70)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.api.python.PairwiseRDD.compute(PythonRDD.scala:342)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)\n",
      "\tat org.apache.spark.scheduler.Task.run(Task.scala:88)\n",
      "\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)\n",
      "\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)\n",
      "\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)\n",
      "\tat java.lang.Thread.run(Thread.java:744)\n",
      "\n",
      "Driver stacktrace:\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1283)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1271)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1270)\n",
      "\tat scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)\n",
      "\tat scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1270)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:697)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:697)\n",
      "\tat scala.Option.foreach(Option.scala:236)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:697)\n",
      "\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1496)\n",
      "\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1458)\n",
      "\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1447)\n",
      "\tat org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:567)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1822)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1835)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1848)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1919)\n",
      "\tat org.apache.spark.rdd.RDD$$anonfun$collect$1.apply(RDD.scala:905)\n",
      "\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:147)\n",
      "\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:108)\n",
      "\tat org.apache.spark.rdd.RDD.withScope(RDD.scala:306)\n",
      "\tat org.apache.spark.rdd.RDD.collect(RDD.scala:904)\n",
      "\tat org.apache.spark.api.python.PythonRDD$.collectAndServe(PythonRDD.scala:405)\n",
      "\tat org.apache.spark.api.python.PythonRDD.collectAndServe(PythonRDD.scala)\n",
      "\tat sun.reflect.GeneratedMethodAccessor55.invoke(Unknown Source)\n",
      "\tat sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\n",
      "\tat java.lang.reflect.Method.invoke(Method.java:606)\n",
      "\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231)\n",
      "\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:379)\n",
      "\tat py4j.Gateway.invoke(Gateway.java:259)\n",
      "\tat py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)\n",
      "\tat py4j.commands.CallCommand.execute(CallCommand.java:79)\n",
      "\tat py4j.GatewayConnection.run(GatewayConnection.java:207)\n",
      "\tat java.lang.Thread.run(Thread.java:744)\n",
      "Caused by: org.apache.spark.api.python.PythonException: Traceback (most recent call last):\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/worker.py\", line 98, in main\n",
      "    command = pickleSer._read_with_length(infile)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/serializers.py\", line 164, in _read_with_length\n",
      "    return self.loads(obj)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/serializers.py\", line 422, in loads\n",
      "    return pickle.loads(obj)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/broadcast.py\", line 39, in _from_id\n",
      "    raise Exception(\"Broadcast variable '%s' not loaded!\" % bid)\n",
      "Exception: (Exception(\"Broadcast variable '0' not loaded!\",), <function _from_id at 0x7f923b0da5f0>, (0L,))\n",
      "\n",
      "\tat org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:166)\n",
      "\tat org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:207)\n",
      "\tat org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:125)\n",
      "\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:70)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:70)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.api.python.PairwiseRDD.compute(PythonRDD.scala:342)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)\n",
      "\tat org.apache.spark.scheduler.Task.run(Task.scala:88)\n",
      "\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)\n",
      "\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)\n",
      "\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)\n",
      "\t... 1 more\n",
      "\n",
      "Traceback (most recent call last):\n",
      "  File \"/usr/local/spark/python/pyspark/streaming/util.py\", line 62, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/usr/local/spark/python/pyspark/streaming/dstream.py\", line 159, in <lambda>\n",
      "    func = lambda t, rdd: old_func(rdd)\n",
      "  File \"<ipython-input-2-21c82552202c>\", line 31, in publish_popular_users\n",
      "    popular_rdd.foreachPartition(publish_partition)\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 766, in foreachPartition\n",
      "    self.mapPartitions(func).count()  # Force evaluation\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 1006, in count\n",
      "    return self.mapPartitions(lambda i: [sum(1 for _ in i)]).sum()\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 997, in sum\n",
      "    return self.mapPartitions(lambda x: [sum(x)]).fold(0, operator.add)\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 871, in fold\n",
      "    vals = self.mapPartitions(func).collect()\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 773, in collect\n",
      "    port = self.ctx._jvm.PythonRDD.collectAndServe(self._jrdd.rdd())\n",
      "  File \"/usr/local/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 538, in __call__\n",
      "    self.target_id, self.name)\n",
      "  File \"/usr/local/spark/python/pyspark/sql/utils.py\", line 36, in deco\n",
      "    return f(*a, **kw)\n",
      "  File \"/usr/local/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py\", line 300, in get_return_value\n",
      "    format(target_id, '.', name), value)\n",
      "Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.\n",
      ": org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 1845.0 failed 1 times, most recent failure: Lost task 0.0 in stage 1845.0 (TID 2527, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/worker.py\", line 98, in main\n",
      "    command = pickleSer._read_with_length(infile)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/serializers.py\", line 164, in _read_with_length\n",
      "    return self.loads(obj)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/serializers.py\", line 422, in loads\n",
      "    return pickle.loads(obj)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/broadcast.py\", line 39, in _from_id\n",
      "    raise Exception(\"Broadcast variable '%s' not loaded!\" % bid)\n",
      "Exception: (Exception(\"Broadcast variable '0' not loaded!\",), <function _from_id at 0x7f923b0da5f0>, (0L,))\n",
      "\n",
      "\tat org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:166)\n",
      "\tat org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:207)\n",
      "\tat org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:125)\n",
      "\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:70)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:70)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.api.python.PairwiseRDD.compute(PythonRDD.scala:342)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)\n",
      "\tat org.apache.spark.scheduler.Task.run(Task.scala:88)\n",
      "\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)\n",
      "\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)\n",
      "\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)\n",
      "\tat java.lang.Thread.run(Thread.java:744)\n",
      "\n",
      "Driver stacktrace:\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1283)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1271)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1270)\n",
      "\tat scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)\n",
      "\tat scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1270)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:697)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:697)\n",
      "\tat scala.Option.foreach(Option.scala:236)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:697)\n",
      "\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1496)\n",
      "\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1458)\n",
      "\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1447)\n",
      "\tat org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:567)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1822)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1835)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1848)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1919)\n",
      "\tat org.apache.spark.rdd.RDD$$anonfun$collect$1.apply(RDD.scala:905)\n",
      "\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:147)\n",
      "\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:108)\n",
      "\tat org.apache.spark.rdd.RDD.withScope(RDD.scala:306)\n",
      "\tat org.apache.spark.rdd.RDD.collect(RDD.scala:904)\n",
      "\tat org.apache.spark.api.python.PythonRDD$.collectAndServe(PythonRDD.scala:405)\n",
      "\tat org.apache.spark.api.python.PythonRDD.collectAndServe(PythonRDD.scala)\n",
      "\tat sun.reflect.GeneratedMethodAccessor55.invoke(Unknown Source)\n",
      "\tat sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\n",
      "\tat java.lang.reflect.Method.invoke(Method.java:606)\n",
      "\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231)\n",
      "\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:379)\n",
      "\tat py4j.Gateway.invoke(Gateway.java:259)\n",
      "\tat py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)\n",
      "\tat py4j.commands.CallCommand.execute(CallCommand.java:79)\n",
      "\tat py4j.GatewayConnection.run(GatewayConnection.java:207)\n",
      "\tat java.lang.Thread.run(Thread.java:744)\n",
      "Caused by: org.apache.spark.api.python.PythonException: Traceback (most recent call last):\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/worker.py\", line 98, in main\n",
      "    command = pickleSer._read_with_length(infile)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/serializers.py\", line 164, in _read_with_length\n",
      "    return self.loads(obj)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/serializers.py\", line 422, in loads\n",
      "    return pickle.loads(obj)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/broadcast.py\", line 39, in _from_id\n",
      "    raise Exception(\"Broadcast variable '%s' not loaded!\" % bid)\n",
      "Exception: (Exception(\"Broadcast variable '0' not loaded!\",), <function _from_id at 0x7f923b0da5f0>, (0L,))\n",
      "\n",
      "\tat org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:166)\n",
      "\tat org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:207)\n",
      "\tat org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:125)\n",
      "\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:70)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:70)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.api.python.PairwiseRDD.compute(PythonRDD.scala:342)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)\n",
      "\tat org.apache.spark.scheduler.Task.run(Task.scala:88)\n",
      "\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)\n",
      "\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)\n",
      "\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)\n",
      "\t... 1 more\n",
      "\n",
      "Traceback (most recent call last):\n",
      "  File \"/usr/local/spark/python/pyspark/streaming/util.py\", line 62, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/usr/local/spark/python/pyspark/streaming/dstream.py\", line 159, in <lambda>\n",
      "    func = lambda t, rdd: old_func(rdd)\n",
      "  File \"<ipython-input-2-21c82552202c>\", line 37, in compute_percentile\n",
      "    digest_partitions).reduce(add).percentile(0.95)\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 799, in reduce\n",
      "    vals = self.mapPartitions(func).collect()\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 773, in collect\n",
      "    port = self.ctx._jvm.PythonRDD.collectAndServe(self._jrdd.rdd())\n",
      "  File \"/usr/local/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 538, in __call__\n",
      "    self.target_id, self.name)\n",
      "  File \"/usr/local/spark/python/pyspark/sql/utils.py\", line 36, in deco\n",
      "    return f(*a, **kw)\n",
      "  File \"/usr/local/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py\", line 300, in get_return_value\n",
      "    format(target_id, '.', name), value)\n",
      "Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.\n",
      ": org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 1850.0 failed 1 times, most recent failure: Lost task 0.0 in stage 1850.0 (TID 2529, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/worker.py\", line 98, in main\n",
      "    command = pickleSer._read_with_length(infile)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/serializers.py\", line 164, in _read_with_length\n",
      "    return self.loads(obj)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/serializers.py\", line 422, in loads\n",
      "    return pickle.loads(obj)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/broadcast.py\", line 39, in _from_id\n",
      "    raise Exception(\"Broadcast variable '%s' not loaded!\" % bid)\n",
      "Exception: (Exception(\"Broadcast variable '0' not loaded!\",), <function _from_id at 0x7f923b0da5f0>, (0L,))\n",
      "\n",
      "\tat org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:166)\n",
      "\tat org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:207)\n",
      "\tat org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:125)\n",
      "\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:70)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:70)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.api.python.PairwiseRDD.compute(PythonRDD.scala:342)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)\n",
      "\tat org.apache.spark.scheduler.Task.run(Task.scala:88)\n",
      "\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)\n",
      "\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)\n",
      "\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)\n",
      "\tat java.lang.Thread.run(Thread.java:744)\n",
      "\n",
      "Driver stacktrace:\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1283)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1271)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1270)\n",
      "\tat scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)\n",
      "\tat scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1270)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:697)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:697)\n",
      "\tat scala.Option.foreach(Option.scala:236)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:697)\n",
      "\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1496)\n",
      "\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1458)\n",
      "\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1447)\n",
      "\tat org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:567)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1822)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1835)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1848)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1919)\n",
      "\tat org.apache.spark.rdd.RDD$$anonfun$collect$1.apply(RDD.scala:905)\n",
      "\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:147)\n",
      "\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:108)\n",
      "\tat org.apache.spark.rdd.RDD.withScope(RDD.scala:306)\n",
      "\tat org.apache.spark.rdd.RDD.collect(RDD.scala:904)\n",
      "\tat org.apache.spark.api.python.PythonRDD$.collectAndServe(PythonRDD.scala:405)\n",
      "\tat org.apache.spark.api.python.PythonRDD.collectAndServe(PythonRDD.scala)\n",
      "\tat sun.reflect.GeneratedMethodAccessor55.invoke(Unknown Source)\n",
      "\tat sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\n",
      "\tat java.lang.reflect.Method.invoke(Method.java:606)\n",
      "\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231)\n",
      "\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:379)\n",
      "\tat py4j.Gateway.invoke(Gateway.java:259)\n",
      "\tat py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)\n",
      "\tat py4j.commands.CallCommand.execute(CallCommand.java:79)\n",
      "\tat py4j.GatewayConnection.run(GatewayConnection.java:207)\n",
      "\tat java.lang.Thread.run(Thread.java:744)\n",
      "Caused by: org.apache.spark.api.python.PythonException: Traceback (most recent call last):\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/worker.py\", line 98, in main\n",
      "    command = pickleSer._read_with_length(infile)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/serializers.py\", line 164, in _read_with_length\n",
      "    return self.loads(obj)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/serializers.py\", line 422, in loads\n",
      "    return pickle.loads(obj)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/broadcast.py\", line 39, in _from_id\n",
      "    raise Exception(\"Broadcast variable '%s' not loaded!\" % bid)\n",
      "Exception: (Exception(\"Broadcast variable '0' not loaded!\",), <function _from_id at 0x7f923b0da5f0>, (0L,))\n",
      "\n",
      "\tat org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:166)\n",
      "\tat org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:207)\n",
      "\tat org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:125)\n",
      "\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:70)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:70)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.api.python.PairwiseRDD.compute(PythonRDD.scala:342)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)\n",
      "\tat org.apache.spark.scheduler.Task.run(Task.scala:88)\n",
      "\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)\n",
      "\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)\n",
      "\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)\n",
      "\t... 1 more\n",
      "\n",
      "Traceback (most recent call last):\n",
      "  File \"/usr/local/spark/python/pyspark/streaming/util.py\", line 62, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/usr/local/spark/python/pyspark/streaming/dstream.py\", line 159, in <lambda>\n",
      "    func = lambda t, rdd: old_func(rdd)\n",
      "  File \"<ipython-input-2-21c82552202c>\", line 31, in publish_popular_users\n",
      "    popular_rdd.foreachPartition(publish_partition)\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 766, in foreachPartition\n",
      "    self.mapPartitions(func).count()  # Force evaluation\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 1006, in count\n",
      "    return self.mapPartitions(lambda i: [sum(1 for _ in i)]).sum()\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 997, in sum\n",
      "    return self.mapPartitions(lambda x: [sum(x)]).fold(0, operator.add)\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 871, in fold\n",
      "    vals = self.mapPartitions(func).collect()\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 773, in collect\n",
      "    port = self.ctx._jvm.PythonRDD.collectAndServe(self._jrdd.rdd())\n",
      "  File \"/usr/local/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 538, in __call__\n",
      "    self.target_id, self.name)\n",
      "  File \"/usr/local/spark/python/pyspark/sql/utils.py\", line 36, in deco\n",
      "    return f(*a, **kw)\n",
      "  File \"/usr/local/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py\", line 300, in get_return_value\n",
      "    format(target_id, '.', name), value)\n",
      "Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.\n",
      ": org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 1855.0 failed 1 times, most recent failure: Lost task 0.0 in stage 1855.0 (TID 2531, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/worker.py\", line 98, in main\n",
      "    command = pickleSer._read_with_length(infile)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/serializers.py\", line 164, in _read_with_length\n",
      "    return self.loads(obj)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/serializers.py\", line 422, in loads\n",
      "    return pickle.loads(obj)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/broadcast.py\", line 39, in _from_id\n",
      "    raise Exception(\"Broadcast variable '%s' not loaded!\" % bid)\n",
      "Exception: (Exception(\"Broadcast variable '0' not loaded!\",), <function _from_id at 0x7f923b0da5f0>, (0L,))\n",
      "\n",
      "\tat org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:166)\n",
      "\tat org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:207)\n",
      "\tat org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:125)\n",
      "\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:70)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:70)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.api.python.PairwiseRDD.compute(PythonRDD.scala:342)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)\n",
      "\tat org.apache.spark.scheduler.Task.run(Task.scala:88)\n",
      "\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)\n",
      "\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)\n",
      "\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)\n",
      "\tat java.lang.Thread.run(Thread.java:744)\n",
      "\n",
      "Driver stacktrace:\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1283)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1271)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1270)\n",
      "\tat scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)\n",
      "\tat scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1270)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:697)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:697)\n",
      "\tat scala.Option.foreach(Option.scala:236)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:697)\n",
      "\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1496)\n",
      "\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1458)\n",
      "\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1447)\n",
      "\tat org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:567)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1822)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1835)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1848)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1919)\n",
      "\tat org.apache.spark.rdd.RDD$$anonfun$collect$1.apply(RDD.scala:905)\n",
      "\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:147)\n",
      "\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:108)\n",
      "\tat org.apache.spark.rdd.RDD.withScope(RDD.scala:306)\n",
      "\tat org.apache.spark.rdd.RDD.collect(RDD.scala:904)\n",
      "\tat org.apache.spark.api.python.PythonRDD$.collectAndServe(PythonRDD.scala:405)\n",
      "\tat org.apache.spark.api.python.PythonRDD.collectAndServe(PythonRDD.scala)\n",
      "\tat sun.reflect.GeneratedMethodAccessor55.invoke(Unknown Source)\n",
      "\tat sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\n",
      "\tat java.lang.reflect.Method.invoke(Method.java:606)\n",
      "\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231)\n",
      "\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:379)\n",
      "\tat py4j.Gateway.invoke(Gateway.java:259)\n",
      "\tat py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)\n",
      "\tat py4j.commands.CallCommand.execute(CallCommand.java:79)\n",
      "\tat py4j.GatewayConnection.run(GatewayConnection.java:207)\n",
      "\tat java.lang.Thread.run(Thread.java:744)\n",
      "Caused by: org.apache.spark.api.python.PythonException: Traceback (most recent call last):\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/worker.py\", line 98, in main\n",
      "    command = pickleSer._read_with_length(infile)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/serializers.py\", line 164, in _read_with_length\n",
      "    return self.loads(obj)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/serializers.py\", line 422, in loads\n",
      "    return pickle.loads(obj)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/broadcast.py\", line 39, in _from_id\n",
      "    raise Exception(\"Broadcast variable '%s' not loaded!\" % bid)\n",
      "Exception: (Exception(\"Broadcast variable '0' not loaded!\",), <function _from_id at 0x7f923b0da5f0>, (0L,))\n",
      "\n",
      "\tat org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:166)\n",
      "\tat org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:207)\n",
      "\tat org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:125)\n",
      "\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:70)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:70)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.api.python.PairwiseRDD.compute(PythonRDD.scala:342)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)\n",
      "\tat org.apache.spark.scheduler.Task.run(Task.scala:88)\n",
      "\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)\n",
      "\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)\n",
      "\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)\n",
      "\t... 1 more\n",
      "\n",
      "Traceback (most recent call last):\n",
      "  File \"/usr/local/spark/python/pyspark/streaming/util.py\", line 62, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/usr/local/spark/python/pyspark/streaming/dstream.py\", line 159, in <lambda>\n",
      "    func = lambda t, rdd: old_func(rdd)\n",
      "  File \"<ipython-input-2-21c82552202c>\", line 37, in compute_percentile\n",
      "    digest_partitions).reduce(add).percentile(0.95)\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 799, in reduce\n",
      "    vals = self.mapPartitions(func).collect()\n",
      "  File \"/usr/local/spark/python/pyspark/rdd.py\", line 773, in collect\n",
      "    port = self.ctx._jvm.PythonRDD.collectAndServe(self._jrdd.rdd())\n",
      "  File \"/usr/local/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 538, in __call__\n",
      "    self.target_id, self.name)\n",
      "  File \"/usr/local/spark/python/pyspark/sql/utils.py\", line 36, in deco\n",
      "    return f(*a, **kw)\n",
      "  File \"/usr/local/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py\", line 300, in get_return_value\n",
      "    format(target_id, '.', name), value)\n",
      "Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.\n",
      ": org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 1861.0 failed 1 times, most recent failure: Lost task 0.0 in stage 1861.0 (TID 2534, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/worker.py\", line 98, in main\n",
      "    command = pickleSer._read_with_length(infile)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/serializers.py\", line 164, in _read_with_length\n",
      "    return self.loads(obj)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/serializers.py\", line 422, in loads\n",
      "    return pickle.loads(obj)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/broadcast.py\", line 39, in _from_id\n",
      "    raise Exception(\"Broadcast variable '%s' not loaded!\" % bid)\n",
      "Exception: (Exception(\"Broadcast variable '0' not loaded!\",), <function _from_id at 0x7f923b0da5f0>, (0L,))\n",
      "\n",
      "\tat org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:166)\n",
      "\tat org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:207)\n",
      "\tat org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:125)\n",
      "\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:70)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:70)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.api.python.PairwiseRDD.compute(PythonRDD.scala:342)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)\n",
      "\tat org.apache.spark.scheduler.Task.run(Task.scala:88)\n",
      "\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)\n",
      "\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)\n",
      "\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)\n",
      "\tat java.lang.Thread.run(Thread.java:744)\n",
      "\n",
      "Driver stacktrace:\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1283)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1271)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1270)\n",
      "\tat scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)\n",
      "\tat scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1270)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:697)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:697)\n",
      "\tat scala.Option.foreach(Option.scala:236)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:697)\n",
      "\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1496)\n",
      "\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1458)\n",
      "\tat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1447)\n",
      "\tat org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)\n",
      "\tat org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:567)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1822)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1835)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1848)\n",
      "\tat org.apache.spark.SparkContext.runJob(SparkContext.scala:1919)\n",
      "\tat org.apache.spark.rdd.RDD$$anonfun$collect$1.apply(RDD.scala:905)\n",
      "\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:147)\n",
      "\tat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:108)\n",
      "\tat org.apache.spark.rdd.RDD.withScope(RDD.scala:306)\n",
      "\tat org.apache.spark.rdd.RDD.collect(RDD.scala:904)\n",
      "\tat org.apache.spark.api.python.PythonRDD$.collectAndServe(PythonRDD.scala:405)\n",
      "\tat org.apache.spark.api.python.PythonRDD.collectAndServe(PythonRDD.scala)\n",
      "\tat sun.reflect.GeneratedMethodAccessor55.invoke(Unknown Source)\n",
      "\tat sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\n",
      "\tat java.lang.reflect.Method.invoke(Method.java:606)\n",
      "\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231)\n",
      "\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:379)\n",
      "\tat py4j.Gateway.invoke(Gateway.java:259)\n",
      "\tat py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)\n",
      "\tat py4j.commands.CallCommand.execute(CallCommand.java:79)\n",
      "\tat py4j.GatewayConnection.run(GatewayConnection.java:207)\n",
      "\tat java.lang.Thread.run(Thread.java:744)\n",
      "Caused by: org.apache.spark.api.python.PythonException: Traceback (most recent call last):\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/worker.py\", line 98, in main\n",
      "    command = pickleSer._read_with_length(infile)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/serializers.py\", line 164, in _read_with_length\n",
      "    return self.loads(obj)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/serializers.py\", line 422, in loads\n",
      "    return pickle.loads(obj)\n",
      "  File \"/usr/local/spark/python/lib/pyspark.zip/pyspark/broadcast.py\", line 39, in _from_id\n",
      "    raise Exception(\"Broadcast variable '%s' not loaded!\" % bid)\n",
      "Exception: (Exception(\"Broadcast variable '0' not loaded!\",), <function _from_id at 0x7f923b0da5f0>, (0L,))\n",
      "\n",
      "\tat org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:166)\n",
      "\tat org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:207)\n",
      "\tat org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:125)\n",
      "\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:70)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:70)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.api.python.PairwiseRDD.compute(PythonRDD.scala:342)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:264)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)\n",
      "\tat org.apache.spark.scheduler.Task.run(Task.scala:88)\n",
      "\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)\n",
      "\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)\n",
      "\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)\n",
      "\t... 1 more\n",
      "\n"
     ]
    },
    {
     "ename": "KeyboardInterrupt",
     "evalue": "",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mKeyboardInterrupt\u001b[0m                         Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-4-85ac66af70bf>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m\n\u001b[0;32m      7\u001b[0m \u001b[0mpopular_stream\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mforeachRDD\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mpublish_popular_users\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      8\u001b[0m \u001b[0mssc\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mstart\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m----> 9\u001b[1;33m \u001b[0mssc\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mawaitTermination\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[1;32m/usr/local/spark/python/pyspark/streaming/context.py\u001b[0m in \u001b[0;36mawaitTermination\u001b[1;34m(self, timeout)\u001b[0m\n\u001b[0;32m    245\u001b[0m         \"\"\"\n\u001b[0;32m    246\u001b[0m         \u001b[1;32mif\u001b[0m \u001b[0mtimeout\u001b[0m \u001b[1;32mis\u001b[0m \u001b[0mNone\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 247\u001b[1;33m             \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jssc\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mawaitTermination\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m    248\u001b[0m         \u001b[1;32melse\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    249\u001b[0m             \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jssc\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mawaitTerminationOrTimeout\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mint\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mtimeout\u001b[0m \u001b[1;33m*\u001b[0m \u001b[1;36m1000\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m/usr/local/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\u001b[0m in \u001b[0;36m__call__\u001b[1;34m(self, *args)\u001b[0m\n\u001b[0;32m    534\u001b[0m             \u001b[0mEND_COMMAND_PART\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    535\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 536\u001b[1;33m         \u001b[0manswer\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mgateway_client\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msend_command\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mcommand\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m    537\u001b[0m         return_value = get_return_value(answer, self.gateway_client,\n\u001b[0;32m    538\u001b[0m                 self.target_id, self.name)\n",
      "\u001b[1;32m/usr/local/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\u001b[0m in \u001b[0;36msend_command\u001b[1;34m(self, command, retry)\u001b[0m\n\u001b[0;32m    362\u001b[0m         \u001b[0mconnection\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_get_connection\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    363\u001b[0m         \u001b[1;32mtry\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 364\u001b[1;33m             \u001b[0mresponse\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mconnection\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msend_command\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mcommand\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m    365\u001b[0m             \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_give_back_connection\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mconnection\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    366\u001b[0m         \u001b[1;32mexcept\u001b[0m \u001b[0mPy4JNetworkError\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m/usr/local/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\u001b[0m in \u001b[0;36msend_command\u001b[1;34m(self, command)\u001b[0m\n\u001b[0;32m    471\u001b[0m         \u001b[1;32mtry\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    472\u001b[0m             \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msocket\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msendall\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mcommand\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mencode\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m'utf-8'\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 473\u001b[1;33m             \u001b[0manswer\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0msmart_decode\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mstream\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mreadline\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m[\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m-\u001b[0m\u001b[1;36m1\u001b[0m\u001b[1;33m]\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m    474\u001b[0m             \u001b[0mlogger\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mdebug\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m\"Answer received: {0}\"\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mformat\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0manswer\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    475\u001b[0m             \u001b[1;31m# Happens when a the other end is dead. There might be an empty\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m/usr/lib/python2.7/socket.pyc\u001b[0m in \u001b[0;36mreadline\u001b[1;34m(self, size)\u001b[0m\n\u001b[0;32m    428\u001b[0m                     \u001b[1;32mtry\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    429\u001b[0m                         \u001b[1;32mwhile\u001b[0m \u001b[0mdata\u001b[0m \u001b[1;33m!=\u001b[0m \u001b[1;34m\"\\n\"\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 430\u001b[1;33m                             \u001b[0mdata\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mrecv\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;36m1\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m    431\u001b[0m                             \u001b[1;32mif\u001b[0m \u001b[1;32mnot\u001b[0m \u001b[0mdata\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    432\u001b[0m                                 \u001b[1;32mbreak\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;31mKeyboardInterrupt\u001b[0m: "
     ]
    }
   ],
   "source": [
    "scores = {'profile.picture.like': 2, 'profile.view': 1, 'message.private': 3}\n",
    "scores_b = sc.broadcast(scores)\n",
    "mapped = kvs.map(load_msg)\n",
    "updated = mapped.updateStateByKey(update_scorecount)\n",
    "updated.foreachRDD(compute_percentile)\n",
    "popular_stream = updated.transform(filter_most_popular)\n",
    "popular_stream.foreachRDD(publish_popular_users)\n",
    "ssc.start()\n",
    "ssc.awaitTermination()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}