bitcoin-analytics.ipynb

{
 "metadata": {
  "name": "",
  "signature": "sha256:b77e9575c3eb322d93479dd36d00c86c03295feba822e02ae884f4f1f109cbff"
 },
 "nbformat": 3,
 "nbformat_minor": 0,
 "worksheets": [
  {
   "cells": [
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "sc"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 1,
       "text": [
        "<pyspark.context.SparkContext at 0x7f8f606ce250>"
       ]
      }
     ],
     "prompt_number": 1
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "from pyspark.sql import HiveContext, Row\n",
      "sqlContext = HiveContext(sc)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 2
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "import block as blk\n",
      "import struct\n",
      "import io"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 3
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "%load_ext autoreload\n",
      "%autoreload 2"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 4
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "import hashlib\n",
      "h1 = hashlib.sha256(\"this is a test\").hexdigest()\n",
      "print(h1)\n",
      "h2 = hashlib.sha256(h1).hexdigest()\n",
      "print(h2)\n"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "2e99758548972a8e8822ad47fa1017ff72f06f3ff6a016851f45c398732bc50c\n",
        "34610a7dc634395a3f5f8b5cbcae0be10604358f04acb8f3fd63a9e9369b83d9\n"
       ]
      }
     ],
     "prompt_number": 5
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "print(sc.version)\n",
      "print(sc.defaultParallelism)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "1.2.0\n",
        "4\n"
       ]
      }
     ],
     "prompt_number": 6
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# compute a block-hash\n",
      "ver = 1       \n",
      "prev_block = \"0000000000000000000000000000000000000000000000000000000000000000\"\n",
      "mrkl_root = \"4a5e1e4baab89f3a32518a88c31bc87f618f76673e2cc77ab2127b7afdeda33b\"\n",
      "time_ = 0x495fab29\n",
      "bits = 0x1d00ffff\n",
      "nonce = 0x7c2bac1d\n",
      "\n",
      "header = ( struct.pack(\"<L\", ver) + prev_block.decode('hex')[::-1] +\n",
      "    mrkl_root.decode('hex')[::-1] + struct.pack(\"<LLL\", time_, bits, nonce))\n",
      "hash = hashlib.sha256(hashlib.sha256(header).digest()).digest() \n",
      "print(hash[::-1].encode('hex'))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "000000000019d6689c085ae165831e934ff763ae46a2a6c172b3f1b60a8ce26f\n"
       ]
      }
     ],
     "prompt_number": 8
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# parse a single .dat file\n",
      "blocks = blk.parseBlockFile(\"/home/jeff/prj/bitcoin-block-data/blk00000.dat\")"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 9
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "blks = sc.binaryFiles(\"/home/jeff/prj/bitcoin-block-data/\")"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 10
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "print(blks.count())"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "2\n"
       ]
      }
     ],
     "prompt_number": 11
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "def printFirstByte(f):\n",
      "    print(f.read(10))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 12
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "print(blks.keys().first())"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "file:/home/jeff/prj/bitcoin-block-data/blk00000.dat\n"
       ]
      }
     ],
     "prompt_number": 13
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "parsed = blks.flatMap(lambda x: blk.parseBlockBytes(io.BytesIO(x[1])))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 14
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "print(parsed.count())"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "Traceback (most recent call last):\n",
        "  File \"/usr/lib/python2.7/SocketServer.py\", line 295, in _handle_request_noblock\n",
        "    self.process_request(request, client_address)\n",
        "  File \"/usr/lib/python2.7/SocketServer.py\", line 321, in process_request\n",
        "    self.finish_request(request, client_address)\n",
        "  File \"/usr/lib/python2.7/SocketServer.py\", line 334, in finish_request\n",
        "    self.RequestHandlerClass(request, client_address, self)\n",
        "  File \"/usr/lib/python2.7/SocketServer.py\", line 649, in __init__\n",
        "    self.handle()\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/accumulators.py\", line 246, in handle\n",
        "    num_updates = read_int(self.rfile)\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/serializers.py\", line 511, in read_int\n",
        "    raise EOFError\n",
        "EOFError\n",
        "ERROR:py4j.java_gateway:Error while sending or receiving.\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 479, in send_command\n",
        "    raise Py4JError(\"Answer from Java side is empty\")\n",
        "Py4JError: Answer from Java side is empty\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server\n",
        "Traceback (most recent call last):\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\", line 425, in start\n",
        "    self.socket.connect((self.address, self.port))\n",
        "  File \"/usr/lib/python2.7/socket.py\", line 224, in meth\n",
        "    return getattr(self._sock,name)(*args)\n",
        "error: [Errno 111] Connection refused\n"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "----------------------------------------\n",
        "Exception happened during processing of request from ('127.0.0.1', 35765)\n",
        "----------------------------------------\n"
       ]
      },
      {
       "ename": "Py4JNetworkError",
       "evalue": "An error occurred while trying to connect to the Java server",
       "output_type": "pyerr",
       "traceback": [
        "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m\n\u001b[1;31mPy4JNetworkError\u001b[0m                          Traceback (most recent call last)",
        "\u001b[1;32m<ipython-input-15-49587a4b379d>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m\n\u001b[1;32m----> 1\u001b[1;33m \u001b[1;32mprint\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mparsed\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mcount\u001b[0m\u001b[1;33m(\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/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/rdd.pyc\u001b[0m in \u001b[0;36mcount\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m    818\u001b[0m         \u001b[1;36m3\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    819\u001b[0m         \"\"\"\n\u001b[1;32m--> 820\u001b[1;33m         \u001b[1;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mmapPartitions\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;32mlambda\u001b[0m \u001b[0mi\u001b[0m\u001b[1;33m:\u001b[0m \u001b[1;33m[\u001b[0m\u001b[0msum\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;36m1\u001b[0m \u001b[1;32mfor\u001b[0m \u001b[0m_\u001b[0m \u001b[1;32min\u001b[0m \u001b[0mi\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m]\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msum\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    821\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    822\u001b[0m     \u001b[1;32mdef\u001b[0m \u001b[0mstats\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
        "\u001b[1;32m/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/rdd.pyc\u001b[0m in \u001b[0;36msum\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m    809\u001b[0m         \u001b[1;36m6.0\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    810\u001b[0m         \"\"\"\n\u001b[1;32m--> 811\u001b[1;33m         \u001b[1;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mmapPartitions\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;32mlambda\u001b[0m \u001b[0mx\u001b[0m\u001b[1;33m:\u001b[0m \u001b[1;33m[\u001b[0m\u001b[0msum\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mx\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m]\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mreduce\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0moperator\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0madd\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m    812\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    813\u001b[0m     \u001b[1;32mdef\u001b[0m \u001b[0mcount\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
        "\u001b[1;32m/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/rdd.pyc\u001b[0m in \u001b[0;36mreduce\u001b[1;34m(self, f)\u001b[0m\n\u001b[0;32m    714\u001b[0m             \u001b[1;32myield\u001b[0m \u001b[0mreduce\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mf\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0miterator\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0minitial\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    715\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 716\u001b[1;33m         \u001b[0mvals\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mmapPartitions\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mfunc\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mcollect\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    717\u001b[0m         \u001b[1;32mif\u001b[0m \u001b[0mvals\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    718\u001b[0m             \u001b[1;32mreturn\u001b[0m \u001b[0mreduce\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mf\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mvals\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
        "\u001b[1;32m/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/rdd.pyc\u001b[0m in \u001b[0;36mcollect\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m    675\u001b[0m         \"\"\"\n\u001b[0;32m    676\u001b[0m         \u001b[1;32mwith\u001b[0m \u001b[0mSCCallSiteSync\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mcontext\u001b[0m\u001b[1;33m)\u001b[0m \u001b[1;32mas\u001b[0m \u001b[0mcss\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 677\u001b[1;33m             \u001b[0mbytesInJava\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jrdd\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mcollect\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0miterator\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    678\u001b[0m         \u001b[1;32mreturn\u001b[0m \u001b[0mlist\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_collect_iterator_through_file\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mbytesInJava\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    679\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n",
        "\u001b[1;32m/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/traceback_utils.pyc\u001b[0m in \u001b[0;36m__exit__\u001b[1;34m(self, type, value, tb)\u001b[0m\n\u001b[0;32m     76\u001b[0m         \u001b[0mSCCallSiteSync\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_spark_stack_depth\u001b[0m \u001b[1;33m-=\u001b[0m \u001b[1;36m1\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     77\u001b[0m         \u001b[1;32mif\u001b[0m \u001b[0mSCCallSiteSync\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_spark_stack_depth\u001b[0m \u001b[1;33m==\u001b[0m \u001b[1;36m0\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 78\u001b[1;33m             \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_context\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jsc\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msetCallSite\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mNone\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m",
        "\u001b[1;32m/home/jeff/spark-1.2.0-bin-hadoop2.4/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/home/jeff/spark-1.2.0-bin-hadoop2.4/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    360\u001b[0m          \u001b[0mthe\u001b[0m \u001b[0mPy4J\u001b[0m \u001b[0mprotocol\u001b[0m\u001b[1;33m.\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    361\u001b[0m         \"\"\"\n\u001b[1;32m--> 362\u001b[1;33m         \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[0m\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[0;32m    364\u001b[0m             \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[1;32m/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\u001b[0m in \u001b[0;36m_get_connection\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m    316\u001b[0m             \u001b[0mconnection\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mdeque\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mpop\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    317\u001b[0m         \u001b[1;32mexcept\u001b[0m \u001b[0mException\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 318\u001b[1;33m             \u001b[0mconnection\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_create_connection\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    319\u001b[0m         \u001b[1;32mreturn\u001b[0m \u001b[0mconnection\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    320\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n",
        "\u001b[1;32m/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\u001b[0m in \u001b[0;36m_create_connection\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m    323\u001b[0m         connection = GatewayConnection(self.address, self.port,\n\u001b[0;32m    324\u001b[0m                 self.auto_close, self.gateway_property)\n\u001b[1;32m--> 325\u001b[1;33m         \u001b[0mconnection\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[0m\u001b[0;32m    326\u001b[0m         \u001b[1;32mreturn\u001b[0m \u001b[0mconnection\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    327\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n",
        "\u001b[1;32m/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py\u001b[0m in \u001b[0;36mstart\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m    430\u001b[0m                 \u001b[1;34m'server'\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    431\u001b[0m             \u001b[0mlogger\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mexception\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mmsg\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 432\u001b[1;33m             \u001b[1;32mraise\u001b[0m \u001b[0mPy4JNetworkError\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mmsg\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m    433\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    434\u001b[0m     \u001b[1;32mdef\u001b[0m \u001b[0mclose\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
        "\u001b[1;31mPy4JNetworkError\u001b[0m: An error occurred while trying to connect to the Java server"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "Exception in thread Thread-3:\n",
        "Traceback (most recent call last):\n",
        "  File \"/usr/lib/python2.7/threading.py\", line 810, in __bootstrap_inner\n",
        "    self.run()\n",
        "  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/java_gateway.py\", line 103, in run\n",
        "    sys.stderr.write(line)\n",
        "  File \"/usr/local/lib/python2.7/dist-packages/IPython/kernel/zmq/iostream.py\", line 205, in write\n",
        "    self._buffer.write(string)\n",
        "ValueError: I/O operation on closed file\n",
        "\n"
       ]
      }
     ],
     "prompt_number": 15
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "parsed.first().printBlock()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "magic_no:\t0xd9b4bef9\n",
        "size:    \t285 bytes\n",
        "Block header:\t\n",
        "\t\tVersion: 1 \n",
        "\t\tPreviousHash: 0000000000000000000000000000000000000000000000000000000000000000 \n",
        "\t\tMerkle: 4a5e1e4baab89f3a32518a88c31bc87f618f76673e2cc77ab2127b7afdeda33b \n",
        "\t\tTime: 495fab29 \n",
        "\t\tBits: 1d00ffff \n",
        "\t\tNonce: 7c2bac1d \n",
        "\t\tPrefix: 0100000000000000000000000000000000000000000000000000000000000000000000003ba3edfd7a7b12b27ac72c3e67768f617fc81bc3888a51323a9fb8aa4b1e5e4a29ab5f49ffff001d1dac2b7c \n",
        "\t\tBlockHash: 000000000019d6689c085ae165831e934ff763ae46a2a6c172b3f1b60a8ce26f \n",
        "\t\t\n",
        "Transactions: \t1\n",
        "==================================================\n",
        " TX NUMBER: 1\n",
        "==================================================\n",
        "Version: 1\n",
        "Inputs count: 1\n",
        "---Inputs---\n",
        "PrevHash: 0000000000000000000000000000000000000000000000000000000000000000 \n",
        "Prev Tx out index: 4294967295 \n",
        "Txin Script Len: 77 \n",
        "scriptSig: 04ffff001d0104455468652054696d65732030332f4a616e2f32303039204368616e63656c6c6f72206f6e206272696e6b206f66207365636f6e64206261696c6f757420666f722062616e6b73 \n",
        "Sequence: ffffffff\n",
        "Outputs count: 1\n",
        "---Outputs---\n",
        "Value (satoshis): 5000000000 (50.000000 btc)\n",
        "Txout Script Len: 67\n",
        "scriptPubKey: 4104678afdb0fe5548271967f1a67130b7105cd6a828e03909a67962e0ea1f61deb649f6bc3f4cef38c4f35504e51ec112de5c384df7ba0b8d578a4c702b6bf11d5fac\n",
        "Lock_time:       0\n",
        "\n",
        "\n"
       ]
      }
     ],
     "prompt_number": 15
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "print(len(blocks))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "119965\n"
       ]
      }
     ],
     "prompt_number": 8
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "blocks[0].printBlock()\n",
      "blocks[100000].printBlock()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "magic_no:\t0xd9b4bef9\n",
        "size:    \t285 bytes\n",
        "Block header:\t\n",
        "\t\tVersion: 1 \n",
        "\t\tPreviousHash: 0000000000000000000000000000000000000000000000000000000000000000 \n",
        "\t\tMerkle: 4a5e1e4baab89f3a32518a88c31bc87f618f76673e2cc77ab2127b7afdeda33b \n",
        "\t\tTime: 495fab29 \n",
        "\t\tBits: 1d00ffff \n",
        "\t\tNonce: 7c2bac1d \n",
        "\t\tPrefix: 0100000000000000000000000000000000000000000000000000000000000000000000003ba3edfd7a7b12b27ac72c3e67768f617fc81bc3888a51323a9fb8aa4b1e5e4a29ab5f49ffff001d1dac2b7c \n",
        "\t\tBlockHash: 000000000019d6689c085ae165831e934ff763ae46a2a6c172b3f1b60a8ce26f \n",
        "\t\t\n",
        "Transactions: \t1\n",
        "==================================================\n",
        " TX NUMBER: 1\n",
        "==================================================\n",
        "Version: 1\n",
        "Inputs count: 1\n",
        "---Inputs---\n",
        "PrevHash: 0000000000000000000000000000000000000000000000000000000000000000 \n",
        "Prev Tx out index: 4294967295 \n",
        "Txin Script Len: 77 \n",
        "scriptSig: 04ffff001d0104455468652054696d65732030332f4a616e2f32303039204368616e63656c6c6f72206f6e206272696e6b206f66207365636f6e64206261696c6f757420666f722062616e6b73 \n",
        "Sequence: ffffffff\n",
        "Outputs count: 1\n",
        "---Outputs---\n",
        "Value (satoshis): 5000000000 (50.000000 btc)\n",
        "Txout Script Len: 67\n",
        "scriptPubKey: 4104678afdb0fe5548271967f1a67130b7105cd6a828e03909a67962e0ea1f61deb649f6bc3f4cef38c4f35504e51ec112de5c384df7ba0b8d578a4c702b6bf11d5fac\n",
        "Lock_time:       0\n",
        "\n",
        "\n",
        "magic_no:\t0xd9b4bef9\n",
        "size:    \t957 bytes\n",
        "Block header:\t\n",
        "\t\tVersion: 1 \n",
        "\t\tPreviousHash: 000000000002d01c1fccc21636b607dfd930d31d01c3a62104612a1719011250 \n",
        "\t\tMerkle: f3e94742aca4b5ef85488dc37c06c3282295ffec960994b2c0d5ac2a25a95766 \n",
        "\t\tTime: 4d1b2237 \n",
        "\t\tBits: 1b04864c \n",
        "\t\tNonce: 10572b0f \n",
        "\t\tPrefix: 0100000050120119172a610421a6c3011dd330d9df07b63616c2cc1f1cd00200000000006657a9252aacd5c0b2940996ecff952228c3067cc38d4885efb5a4ac4247e9f337221b4d4c86041b0f2b5710 \n",
        "\t\tBlockHash: 000000000003ba27aa200b1cecaad478d2b00432346c3f1f3986da1afd33e506 \n",
        "\t\t\n",
        "Transactions: \t4\n",
        "==================================================\n",
        " TX NUMBER: 1\n",
        "==================================================\n",
        "Version: 1\n",
        "Inputs count: 1\n",
        "---Inputs---\n",
        "PrevHash: 0000000000000000000000000000000000000000000000000000000000000000 \n",
        "Prev Tx out index: 4294967295 \n",
        "Txin Script Len: 8 \n",
        "scriptSig: 044c86041b020602 \n",
        "Sequence: ffffffff\n",
        "Outputs count: 1\n",
        "---Outputs---\n",
        "Value (satoshis): 5000000000 (50.000000 btc)\n",
        "Txout Script Len: 67\n",
        "scriptPubKey: 41041b0e8c2567c12536aa13357b79a073dc4444acb83c4ec7a0e2f99dd7457516c5817242da796924ca4e99947d087fedf9ce467cb9f7c6287078f801df276fdf84ac\n",
        "Lock_time:       0\n",
        "\n",
        "\n",
        "==================================================\n",
        " TX NUMBER: 2\n",
        "==================================================\n",
        "Version: 1\n",
        "Inputs count: 1\n",
        "---Inputs---\n",
        "PrevHash: 87a157f3fd88ac7907c05fc55e271dc4acdc5605d187d646604ca8c0e9382e03 \n",
        "Prev Tx out index: 0 \n",
        "Txin Script Len: 140 \n",
        "scriptSig: 493046022100c352d3dd993a981beba4a63ad15c209275ca9470abfcd57da93b58e4eb5dce82022100840792bc1f456062819f15d33ee7055cf7b5ee1af1ebcc6028d9cdb1c3af7748014104f46db5e9d61a9dc27b8d64ad23e7383a4e6ca164593c2527c038c0857eb67ee8e825dca65046b82c9331586c82e0fd1f633f25f87c161bc6f8a630121df2b3d3 \n",
        "Sequence: ffffffff\n",
        "Outputs count: 2\n",
        "---Outputs---\n",
        "Value (satoshis): 556000000 (5.560000 btc)\n",
        "Txout Script Len: 25\n",
        "scriptPubKey: 76a914c398efa9c392ba6013c5e04ee729755ef7f58b3288ac\n",
        "Value (satoshis): 4444000000 (44.440000 btc)\n",
        "Txout Script Len: 25\n",
        "scriptPubKey: 76a914948c765a6914d43f2a7ac177da2c2f6b52de3d7c88ac\n",
        "Lock_time:       0\n",
        "\n",
        "\n",
        "==================================================\n",
        " TX NUMBER: 3\n",
        "==================================================\n",
        "Version: 1\n",
        "Inputs count: 1\n",
        "---Inputs---\n",
        "PrevHash: cf4e2978d0611ce46592e02d7e7daf8627a316ab69759a9f3df109a7f2bf3ec3 \n",
        "Prev Tx out index: 1 \n",
        "Txin Script Len: 138 \n",
        "scriptSig: 4730440220032d30df5ee6f57fa46cddb5eb8d0d9fe8de6b342d27942ae90a3231e0ba333e02203deee8060fdc70230a7f5b4ad7d7bc3e628cbe219a886b84269eaeb81e26b4fe014104ae31c31bf91278d99b8377a35bbce5b27d9fff15456839e919453fc7b3f721f0ba403ff96c9deeb680e5fd341c0fc3a7b90da4631ee39560639db462e9cb850f \n",
        "Sequence: ffffffff\n",
        "Outputs count: 2\n",
        "---Outputs---\n",
        "Value (satoshis): 1000000 (0.010000 btc)\n",
        "Txout Script Len: 25\n",
        "scriptPubKey: 76a914b0dcbf97eabf4404e31d952477ce822dadbe7e1088ac\n",
        "Value (satoshis): 299000000 (2.990000 btc)\n",
        "Txout Script Len: 25\n",
        "scriptPubKey: 76a9146b1281eec25ab4e1e0793ff4e08ab1abb3409cd988ac\n",
        "Lock_time:       0\n",
        "\n",
        "\n",
        "==================================================\n",
        " TX NUMBER: 4\n",
        "==================================================\n",
        "Version: 1\n",
        "Inputs count: 1\n",
        "---Inputs---\n",
        "PrevHash: f4515fed3dc4a19b90a317b9840c243bac26114cf637522373a7d486b372600b \n",
        "Prev Tx out index: 0 \n",
        "Txin Script Len: 140 \n",
        "scriptSig: 493046022100bb1ad26df930a51cce110cf44f7a48c3c561fd977500b1ae5d6b6fd13d0b3f4a022100c5b42951acedff14abba2736fd574bdb465f3e6f8da12e2c5303954aca7f78f3014104a7135bfe824c97ecc01ec7d7e336185c81e2aa2c41ab175407c09484ce9694b44953fcb751206564a9c24dd094d42fdbfdd5aad3e063ce6af4cfaaea4ea14fbb \n",
        "Sequence: ffffffff\n",
        "Outputs count: 1\n",
        "---Outputs---\n",
        "Value (satoshis): 1000000 (0.010000 btc)\n",
        "Txout Script Len: 25\n",
        "scriptPubKey: 76a91439aa3d569e06a1d7926dc4be1193c99bf2eb9ee088ac\n",
        "Lock_time:       0\n",
        "\n",
        "\n"
       ]
      }
     ],
     "prompt_number": 9
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# create a spark partition for each .dat file\n",
      "files = [\"blk00000.dat\",\"blk00001.dat\"]\n",
      "blockFileNames = sc.parallelize(files,len(files))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 7
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "blocksRDD = blockFileNames.flatMap(lambda x: blk.parseBlockFile(x)). \\\n",
      "                keyBy(lambda x: (x.getBlockHash(),x.getBlockPrevHash(),x.getBlockDifficulty())).repartition(4)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 8
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "print(blockFileNames.glom().collect())"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "[['blk00000.dat'], ['blk00001.dat']]\n"
       ]
      }
     ],
     "prompt_number": 9
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "print(blocksRDD.count())"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "131237\n"
       ]
      }
     ],
     "prompt_number": 10
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "tmp = blocksRDD.take(2)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 11
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "print(tmp)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "[(('00000000839a8e6886ab5951d76f411475428afc90947ee320161bbf18eb6048', '000000000019d6689c085ae165831e934ff763ae46a2a6c172b3f1b60a8ce26f', 486604799), <block.Block object at 0x7effa8571690>), (('000000006a625f06636b8bb6ac7b960a8d03705d1ace08b1a19da3fdcc99ddbd', '00000000839a8e6886ab5951d76f411475428afc90947ee320161bbf18eb6048', 486604799), <block.Block object at 0x7effa85717d0>)]\n"
       ]
      }
     ],
     "prompt_number": 12
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "headers = blocksRDD.keys().collect()  # returns dict{key = blockhash, value = prev-blockhash}"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 13
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "from collections import defaultdict\n",
      "headersMap = defaultdict(list)\n",
      "for h in headers:\n",
      "    headersMap[h[1]].append((h[0],h[2]))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 14
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "len(headers)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 15,
       "text": [
        "131237"
       ]
      }
     ],
     "prompt_number": 15
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "len(set(headersMap.keys()))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 16,
       "text": [
        "131237"
       ]
      }
     ],
     "prompt_number": 16
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# set the genesis header hash\n",
      "genesisPrev = '0000000000000000000000000000000000000000000000000000000000000000'\n",
      "genesisHash = '000000000019d6689c085ae165831e934ff763ae46a2a6c172b3f1b60a8ce26f'\n",
      "genesisDifficulty = 0x1d00ffff"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 17
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "import graph_tool as gt\n",
      "import graph_tool.util as gtutil"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 18
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# construct the header graph\n",
      "headerChain = gt.Graph() # key = block hash, value = list of conneced nodes\n",
      "v_hash = headerChain.new_vertex_property(\"string\")\n",
      "v_difficulty = headerChain.new_vertex_property(\"int\")\n",
      "curHash = genesisHash # the header hash we are looking for in prevHash\n",
      "\n",
      "# add the vertex\n",
      "v = headerChain.add_vertex()\n",
      "v_hash[v] = curHash\n",
      "v_difficulty[v] = genesisDifficulty\n",
      "\n",
      "\n",
      "def generateBlockGraph(vertex,graph,vhashes,vdifficulty,blockMap):\n",
      "    # uses depth-first search to construct graph\n",
      "    vList = []\n",
      "    discovered = set()\n",
      "    vList.append(vertex)\n",
      "    while (len(vList) > 0):\n",
      "        vertex = vList.pop()\n",
      "        if (vertex not in discovered): # only process the vertex if is has not been discovered\n",
      "            discovered.add(vertex);\n",
      "            # find blocks above this one and connect them up\n",
      "            if blockMap.has_key(vhashes[vertex]):\n",
      "                nextHeaders = blockMap[vhashes[vertex]] # returns a list of connected headers\n",
      "                for h in nextHeaders:\n",
      "                    v = graph.add_vertex() # add vertex\n",
      "                    vhashes[v] = h[0]  # set the property\n",
      "                    vdifficulty[v] = h[1]\n",
      "                    graph.add_edge(vertex,v) # add the edge\n",
      "                    vList.append(v)\n",
      "    return (graph, vhashes, vdifficulty)\n",
      "\n",
      "res = generateBlockGraph(v, headerChain, v_hash, v_difficulty, headersMap)\n",
      "headerChain = res[0]\n",
      "v_hash = res[1]\n",
      "v_difficulty = res[2]"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 19
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# plot the graph (takes too long!)\n",
      "#gd.graph_draw(headerChain, vertex_text=g.vertex_index,vertex_font_size=8, output_size=(200, 200), output=\"block-chain.png\")"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 20
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "print(res)\n",
      "print(v_hash[headerChain.vertex(0)])\n",
      "print(v_hash[headerChain.vertex(1)])\n",
      "print(v_difficulty[headerChain.vertex(0)])\n",
      "print(v_difficulty[headerChain.vertex(1)])"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "(<Graph object, directed, with 131237 vertices and 131236 edges at 0x7eff88584cd0>, <PropertyMap object with key type 'Vertex' and value type 'string', for Graph 0x7eff88584cd0, at 0x7eff88584c90>, <PropertyMap object with key type 'Vertex' and value type 'int32_t', for Graph 0x7eff88584cd0, at 0x7eff88584d50, with values:\n",
        "[486604799 486604799 486604799 ..., 437461381 437461381 437461381]>)\n",
        "000000000019d6689c085ae165831e934ff763ae46a2a6c172b3f1b60a8ce26f\n",
        "00000000839a8e6886ab5951d76f411475428afc90947ee320161bbf18eb6048\n",
        "486604799\n",
        "486604799\n"
       ]
      }
     ],
     "prompt_number": 21
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# get the leaf nodes\n",
      "headerChain.set_vertex_filter(None)\n",
      "v_leafs = headerChain.new_vertex_property(\"bool\")\n",
      "leafs = []\n",
      "for v in headerChain.vertices():\n",
      "    if (v.out_degree() == 0):\n",
      "        v_leafs[v] = True\n",
      "        leafs.append(v)\n",
      "    else:\n",
      "        v_leafs[v] = False\n",
      "        \n",
      "print([v_hash[v] for v in leafs])"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "['0000000000000177ca06002a5d22214a0ba3122da467ffd7d70e2f61836e4198']\n"
       ]
      }
     ],
     "prompt_number": 22
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# find longest chain\n",
      "# TO DO: compute length based on difficulty instead of number of blocks\n",
      "headerChain.set_vertex_filter(None)\n",
      "blockCount = []\n",
      "for leaf in leafs:\n",
      "    # count the number of verticies in this chain\n",
      "    count = 0\n",
      "    v = leaf\n",
      "    while (v.in_degree() > 0):\n",
      "        count = count + 1\n",
      "        v = v.in_neighbours().next() # relies on the fact that nodes have only one input\n",
      "    blockCount.append(count)\n",
      "    \n",
      "print(\"blockCounts: \" + str(blockCount))\n",
      "mainLeaf = leafs[blockCount.index(max(blockCount))]\n",
      "print(\"main leaf index: \" + str(mainLeaf))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "blockCounts: [131236]\n",
        "main leaf index: 131236\n"
       ]
      }
     ],
     "prompt_number": 23
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# make a propety map for the main block chain\n",
      "on_chain = headerChain.new_vertex_property(\"bool\",vals=False)\n",
      "v = mainLeaf\n",
      "on_chain[v] = True\n",
      "while (v.in_degree() > 0):\n",
      "    v = v.in_neighbours().next() # relies on the fact that nodes have only one input\n",
      "    on_chain[v] = True"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 24
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# assign block numbers\n",
      "block_index = headerChain.new_vertex_property(\"int\",vals = -1)\n",
      "headerChain.set_vertex_filter(on_chain)\n",
      "count = 0\n",
      "v = gtutil.find_vertex(headerChain,v_hash,genesisHash)[0]\n",
      "while (v.out_degree() > 0):\n",
      "    block_index[v] = count  #first block has index 0\n",
      "    count = count + 1\n",
      "    v = v.out_neighbours().next() # relies on the fact that nodes have only one input    "
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 25
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# print the first 25 block headers with block index\n",
      "v = gtutil.find_vertex(headerChain,v_hash,genesisHash)[0]\n",
      "count = 0\n",
      "while (count < 25):\n",
      "    curHash = v_hash[v]\n",
      "    blkIdx = block_index[v]\n",
      "    print(\"Block Hash: \" + str(curHash) + \" Block Index: \" + str(blkIdx))\n",
      "    count = count + 1\n",
      "    v = v.out_neighbours().next() # relies on the fact that nodes have only one input  "
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "Block Hash: 000000000019d6689c085ae165831e934ff763ae46a2a6c172b3f1b60a8ce26f Block Index: 0\n",
        "Block Hash: 00000000839a8e6886ab5951d76f411475428afc90947ee320161bbf18eb6048 Block Index: 1\n",
        "Block Hash: 000000006a625f06636b8bb6ac7b960a8d03705d1ace08b1a19da3fdcc99ddbd Block Index: 2\n",
        "Block Hash: 0000000082b5015589a3fdf2d4baff403e6f0be035a5d9742c1cae6295464449 Block Index: 3\n",
        "Block Hash: 000000004ebadb55ee9096c9a2f8880e09da59c0d68b1c228da88e48844a1485 Block Index: 4\n",
        "Block Hash: 000000009b7262315dbf071787ad3656097b892abffd1f95a1a022f896f533fc Block Index: 5\n",
        "Block Hash: 000000003031a0e73735690c5a1ff2a4be82553b2a12b776fbd3a215dc8f778d Block Index: 6\n",
        "Block Hash: 0000000071966c2b1d065fd446b1e485b2c9d9594acd2007ccbd5441cfc89444 Block Index: 7\n",
        "Block Hash: 00000000408c48f847aa786c2268fc3e6ec2af68e8468a34a28c61b7f1de0dc6 Block Index: 8\n",
        "Block Hash: 000000008d9dc510f23c2657fc4f67bea30078cc05a90eb89e84cc475c080805 Block Index: 9\n",
        "Block Hash: 000000002c05cc2e78923c34df87fd108b22221ac6076c18f3ade378a4d915e9 Block Index: 10\n",
        "Block Hash: 0000000097be56d606cdd9c54b04d4747e957d3608abe69198c661f2add73073 Block Index: 11\n",
        "Block Hash: 0000000027c2488e2510d1acf4369787784fa20ee084c258b58d9fbd43802b5e Block Index: 12\n",
        "Block Hash: 000000005c51de2031a895adc145ee2242e919a01c6d61fb222a54a54b4d3089 Block Index: 13\n",
        "Block Hash: 0000000080f17a0c5a67f663a9bc9969eb37e81666d9321125f0e293656f8a37 Block Index: 14\n",
        "Block Hash: 00000000b3322c8c3ef7d2cf6da009a776e6a99ee65ec5a32f3f345712238473 Block Index: 15\n",
        "Block Hash: 00000000174a25bb399b009cc8deff1c4b3ea84df7e93affaaf60dc3416cc4f5 Block Index: 16\n",
        "Block Hash: 000000003ff1d0d70147acfbef5d6a87460ff5bcfce807c2d5b6f0a66bfdf809 Block Index: 17\n",
        "Block Hash: 000000008693e98cf893e4c85a446b410bb4dfa129bd1be582c09ed3f0261116 Block Index: 18\n",
        "Block Hash: 00000000841cb802ca97cf20fb9470480cae9e5daa5d06b4a18ae2d5dd7f186f Block Index: 19\n",
        "Block Hash: 0000000067a97a2a37b8f190a17f0221e9c3f4fa824ddffdc2e205eae834c8d7 Block Index: 20\n",
        "Block Hash: 000000006f016342d1275be946166cff975c8b27542de70a7113ac6d1ef3294f Block Index: 21\n",
        "Block Hash: 0000000098b58d427a10c860335a21c1a9a7639e96c3d6f1a03d8c8c885b5e3b Block Index: 22\n",
        "Block Hash: 000000000cd339982e556dfffa9de94744a4135c53eeef15b7bcc9bdeb9c2182 Block Index: 23\n",
        "Block Hash: 00000000fc051fbbce89a487e811a5d4319d209785ea4f4b27fc83770d1e415f Block Index: 24\n"
       ]
      }
     ],
     "prompt_number": 26
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# create a dict with key = headerHash and value = (blockIdx, on_chain)\n",
      "headerChain.set_vertex_filter(None)\n",
      "blockDict = {}\n",
      "for v in headerChain.vertices():\n",
      "    blockDict[v_hash[v]] = (block_index[v], on_chain[v])"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 27
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# define a map function to insert block index and on_chain into RDD\n",
      "def genAddMeta(metaDict):\n",
      "    '''constructs a closure to use as a mapper function'''\n",
      "    \n",
      "    def addMeta(el):\n",
      "        blkHash = el[0][0]\n",
      "        meta = metaDict[blkHash]\n",
      "        newEl = list(el)\n",
      "        newEl.append(meta[0])  #block_index\n",
      "        newEl.append(meta[1])  #on_chain\n",
      "        return tuple(newEl)\n",
      "    \n",
      "    return addMeta"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 28
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# add meta data to the RDD\n",
      "addMetaFn = genAddMeta(blockDict)\n",
      "blocksRDDv2 = blocksRDD.map(addMetaFn)\n",
      "blocksRDDv2.count()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "----init: sc._temp_dir ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e-----------------\n",
        "----init: f ----<open file '<fdopen>', mode 'w+b' at 0x7eff8d74ef60>-----------------\n",
        "----init: _path ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e/tmp4NGuAd-----------------\n",
        "----unpersist: self._path: ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e/tmp4NGuAd-----------------"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "\n"
       ]
      },
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 29,
       "text": [
        "131237"
       ]
      }
     ],
     "prompt_number": 29
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "tmp = blocksRDDv2.take(5)\n",
      "print(tmp)\n",
      "print(tmp[0][0])\n",
      "print(tmp[0][1])\n",
      "print(tmp[0][2])"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "----init: sc._temp_dir ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e-----------------\n",
        "----init: f ----<open file '<fdopen>', mode 'w+b' at 0x7eff8d74eb70>-----------------\n",
        "----init: _path ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e/tmpgT186p-----------------\n",
        "----init: sc._temp_dir ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e-----------------"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "\n",
        "----init: f ----<open file '<fdopen>', mode 'w+b' at 0x7eff8d74eb70>-----------------\n",
        "----init: _path ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e/tmpjgzXzB-----------------\n",
        "----unpersist: self._path: ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e/tmpjgzXzB-----------------"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "\n",
        "[(('00000000839a8e6886ab5951d76f411475428afc90947ee320161bbf18eb6048', '000000000019d6689c085ae165831e934ff763ae46a2a6c172b3f1b60a8ce26f', 486604799), <block.Block object at 0x7eff885849d0>, 1, 1), (('000000006a625f06636b8bb6ac7b960a8d03705d1ace08b1a19da3fdcc99ddbd', '00000000839a8e6886ab5951d76f411475428afc90947ee320161bbf18eb6048', 486604799), <block.Block object at 0x7eff88584bd0>, 2, 1), (('000000009700ff3494f215c412cd8c0ceabf1deb0df03ce39bcfc223b769d3c4', '00000000bc919cfb64f62de736d55cf79e3d535b474ace256b4fbb56073f64db', 486604799), <block.Block object at 0x7eff88584dd0>, 31, 1), (('00000000e5cb7c6c273547b0c9421b01e23310ed83f934b96270f35a4d66f6e3', '000000009700ff3494f215c412cd8c0ceabf1deb0df03ce39bcfc223b769d3c4', 486604799), <block.Block object at 0x7eff88584f50>, 32, 1), (('00000000a87073ea3d7af299e02a434598b9c92094afa552e0711afcc0857962', '00000000e5cb7c6c273547b0c9421b01e23310ed83f934b96270f35a4d66f6e3', 486604799), <block.Block object at 0x7eff8522d0d0>, 33, 1)]\n",
        "('00000000839a8e6886ab5951d76f411475428afc90947ee320161bbf18eb6048', '000000000019d6689c085ae165831e934ff763ae46a2a6c172b3f1b60a8ce26f', 486604799)\n",
        "<block.Block object at 0x7eff885849d0>\n",
        "1\n"
       ]
      }
     ],
     "prompt_number": 30
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "blocksRDDv2.getNumPartitions()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 31,
       "text": [
        "4"
       ]
      }
     ],
     "prompt_number": 31
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "blocksRDDv2.count()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "----init: sc._temp_dir ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e-----------------\n",
        "----init: f ----<open file '<fdopen>', mode 'w+b' at 0x7eff8d74ef60>-----------------\n",
        "----init: _path ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e/tmpmHUUSF-----------------\n",
        "----unpersist: self._path: ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e/tmpmHUUSF-----------------"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "\n"
       ]
      },
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 32,
       "text": [
        "131237"
       ]
      }
     ],
     "prompt_number": 32
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "%%bash\n",
      "rm -r blocksRDDv2.txt"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 33
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "blocksRDDv2.saveAsPickleFile('blocksRDDv2.txt')"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 34
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "blocksRDDv3 = sc.pickleFile('blocksRDDv2.txt')"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 35
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "blocksRDDv3.count()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 36,
       "text": [
        "131237"
       ]
      }
     ],
     "prompt_number": 36
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "tmp = blocksRDDv3.take(2)\n",
      "print(tmp[0][0])\n",
      "print(tmp[1][1])"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "('00000000839a8e6886ab5951d76f411475428afc90947ee320161bbf18eb6048', '000000000019d6689c085ae165831e934ff763ae46a2a6c172b3f1b60a8ce26f', 486604799)\n",
        "<block.Block object at 0x7eff8522df90>\n"
       ]
      }
     ],
     "prompt_number": 37
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# convert to schema RDD\n",
      "def toTxs(x):\n",
      "    '''Return a list of transactions for each input block'''\n",
      "    txs = []\n",
      "    blk = x[1]\n",
      "    numTxs = blk.getNumTxs()\n",
      "    for i in range(numTxs):\n",
      "        currentDict = {}\n",
      "        currentDict = blk.updateTxDict(i,currentDict)\n",
      "        currentDict['block_index'] = x[2]\n",
      "        currentDict['on_chain'] = x[3]\n",
      "        txs.append(Row(**currentDict))\n",
      "    return txs\n",
      "\n",
      "txRDD = blocksRDDv2.flatMap(lambda x: toTxs(x))\n",
      "txRDD.count()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "----init: sc._temp_dir ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e-----------------\n",
        "----init: f ----<open file '<fdopen>', mode 'w+b' at 0x7eff8d74ea50>-----------------\n",
        "----init: _path ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e/tmpQ73H3j-----------------\n",
        "----unpersist: self._path: ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e/tmpQ73H3j-----------------"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "\n"
       ]
      },
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 38,
       "text": [
        "754445"
       ]
      }
     ],
     "prompt_number": 38
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "%%bash\n",
      "rm -r txRDD.txt"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 39
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "txRDD.saveAsPickleFile(\"txRDD.txt\")"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "----init: sc._temp_dir ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e-----------------\n",
        "----init: f ----<open file '<fdopen>', mode 'w+b' at 0x7eff9463d150>-----------------\n",
        "----init: _path ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e/tmpH_jzhc-----------------\n"
       ]
      }
     ],
     "prompt_number": 40
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "schTxRDD = sqlContext.inferSchema(txRDD)\n",
      "schTxRDD.registerTempTable(\"schTxRDD\")"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "----init: sc._temp_dir ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e-----------------\n",
        "----init: f ----<open file '<fdopen>', mode 'w+b' at 0x7eff9463d150>-----------------\n",
        "----init: _path ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e/tmp78xKoe-----------------\n",
        "----unpersist: self._path: ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e/tmp78xKoe-----------------"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "\n",
        "----init: sc._temp_dir ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e-----------------"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "\n",
        "----init: f ----<open file '<fdopen>', mode 'w+b' at 0x7eff9463d150>-----------------\n",
        "----init: _path ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e/tmpZgY5gE-----------------\n",
        "----init: sc._temp_dir ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e-----------------"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "\n",
        "----init: f ----<open file '<fdopen>', mode 'w+b' at 0x7eff9463d150>-----------------\n",
        "----init: _path ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e/tmpjhQOMJ-----------------\n",
        "----unpersist: self._path: ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e/tmpjhQOMJ-----------------"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "\n",
        "----init: sc._temp_dir ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e-----------------"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "\n",
        "----init: f ----<open file '<fdopen>', mode 'w+b' at 0x7eff9463d150>-----------------\n",
        "----init: _path ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e/tmpfilSxo-----------------\n",
        "----unpersist: self._path: ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e/tmpfilSxo-----------------"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "\n",
        "----unpersist: self._path: ----/tmp/spark-2f62ae62-7af9-40fd-a0ff-0e4c02ceb74e/tmpZgY5gE-----------------\n"
       ]
      }
     ],
     "prompt_number": 41
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "schTxRDD.getNumPartitions()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 42,
       "text": [
        "4"
       ]
      }
     ],
     "prompt_number": 42
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "%%bash\n",
      "rm -r 'schTxRDD.txt'"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stderr",
       "text": [
        "rm: cannot remove \u2018schTxRDD.txt\u2019: No such file or directory\n"
       ]
      }
     ],
     "prompt_number": 43
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "schTxRDD.getStorageLevel()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 44,
       "text": [
        "StorageLevel(False, False, False, False, 1)"
       ]
      }
     ],
     "prompt_number": 44
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "schTxRDD.count()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "ename": "Py4JJavaError",
       "evalue": "An error occurred while calling o174.count.\n: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 26.0 failed 1 times, most recent failure: Lost task 0.0 in stage 26.0 (TID 52, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):\n  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/worker.py\", line 99, in main\n    command = pickleSer.loads(command.value)\n  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/broadcast.py\", line 112, in value\n    self._value = self.load(self._path)\n  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/broadcast.py\", line 92, in load\n    with open(path, 'rb', 1 << 20) as f:\nIOError: [Errno 2] No such file or directory: '/home/jeff/spark-scratch/spark-7fe17099-83e8-48f6-b46a-8925b025c395/tmpINRFSG'\n\n\tat org.apache.spark.api.python.PythonRDD$$anon$1.read(PythonRDD.scala:137)\n\tat org.apache.spark.api.python.PythonRDD$$anon$1.<init>(PythonRDD.scala:174)\n\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:96)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:68)\n\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)\n\tat org.apache.spark.scheduler.Task.run(Task.scala:56)\n\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196)\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:745)\n\nDriver stacktrace:\n\tat org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1214)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1203)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1202)\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:1202)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:696)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:696)\n\tat scala.Option.foreach(Option.scala:236)\n\tat org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:696)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessActor$$anonfun$receive$2.applyOrElse(DAGScheduler.scala:1420)\n\tat akka.actor.Actor$class.aroundReceive(Actor.scala:465)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessActor.aroundReceive(DAGScheduler.scala:1375)\n\tat akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)\n\tat akka.actor.ActorCell.invoke(ActorCell.scala:487)\n\tat akka.dispatch.Mailbox.processMailbox(Mailbox.scala:238)\n\tat akka.dispatch.Mailbox.run(Mailbox.scala:220)\n\tat akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:393)\n\tat scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)\n\tat scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)\n\tat scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)\n\tat scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)\n",
       "output_type": "pyerr",
       "traceback": [
        "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m\n\u001b[1;31mPy4JJavaError\u001b[0m                             Traceback (most recent call last)",
        "\u001b[1;32m<ipython-input-45-4d609412cd08>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m\n\u001b[1;32m----> 1\u001b[1;33m \u001b[0mschTxRDD\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mcount\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/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/sql.pyc\u001b[0m in \u001b[0;36mcount\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m   1959\u001b[0m         \u001b[0mTrue\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1960\u001b[0m         \"\"\"\n\u001b[1;32m-> 1961\u001b[1;33m         \u001b[1;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jschema_rdd\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mcount\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   1962\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1963\u001b[0m     \u001b[1;32mdef\u001b[0m \u001b[0mcollect\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
        "\u001b[1;32m/home/jeff/spark-1.2.0-bin-hadoop2.4/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    536\u001b[0m         \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[0;32m    537\u001b[0m         return_value = get_return_value(answer, self.gateway_client,\n\u001b[1;32m--> 538\u001b[1;33m                 self.target_id, self.name)\n\u001b[0m\u001b[0;32m    539\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    540\u001b[0m         \u001b[1;32mfor\u001b[0m \u001b[0mtemp_arg\u001b[0m \u001b[1;32min\u001b[0m \u001b[0mtemp_args\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
        "\u001b[1;32m/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py\u001b[0m in \u001b[0;36mget_return_value\u001b[1;34m(answer, gateway_client, target_id, name)\u001b[0m\n\u001b[0;32m    298\u001b[0m                 raise Py4JJavaError(\n\u001b[0;32m    299\u001b[0m                     \u001b[1;34m'An error occurred while calling {0}{1}{2}.\\n'\u001b[0m\u001b[1;33m.\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 300\u001b[1;33m                     format(target_id, '.', name), value)\n\u001b[0m\u001b[0;32m    301\u001b[0m             \u001b[1;32melse\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    302\u001b[0m                 raise Py4JError(\n",
        "\u001b[1;31mPy4JJavaError\u001b[0m: An error occurred while calling o174.count.\n: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 26.0 failed 1 times, most recent failure: Lost task 0.0 in stage 26.0 (TID 52, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):\n  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/worker.py\", line 99, in main\n    command = pickleSer.loads(command.value)\n  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/broadcast.py\", line 112, in value\n    self._value = self.load(self._path)\n  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/broadcast.py\", line 92, in load\n    with open(path, 'rb', 1 << 20) as f:\nIOError: [Errno 2] No such file or directory: '/home/jeff/spark-scratch/spark-7fe17099-83e8-48f6-b46a-8925b025c395/tmpINRFSG'\n\n\tat org.apache.spark.api.python.PythonRDD$$anon$1.read(PythonRDD.scala:137)\n\tat org.apache.spark.api.python.PythonRDD$$anon$1.<init>(PythonRDD.scala:174)\n\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:96)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:68)\n\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)\n\tat org.apache.spark.scheduler.Task.run(Task.scala:56)\n\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196)\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:745)\n\nDriver stacktrace:\n\tat org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1214)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1203)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1202)\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:1202)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:696)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:696)\n\tat scala.Option.foreach(Option.scala:236)\n\tat org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:696)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessActor$$anonfun$receive$2.applyOrElse(DAGScheduler.scala:1420)\n\tat akka.actor.Actor$class.aroundReceive(Actor.scala:465)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessActor.aroundReceive(DAGScheduler.scala:1375)\n\tat akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)\n\tat akka.actor.ActorCell.invoke(ActorCell.scala:487)\n\tat akka.dispatch.Mailbox.processMailbox(Mailbox.scala:238)\n\tat akka.dispatch.Mailbox.run(Mailbox.scala:220)\n\tat akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:393)\n\tat scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)\n\tat scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)\n\tat scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)\n\tat scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)\n"
       ]
      }
     ],
     "prompt_number": 45
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "print(schTxRDD.schemaString())"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "root\n",
        " |-- bits: integer (nullable = true)\n",
        " |-- block_index: integer (nullable = true)\n",
        " |-- blockhash: string (nullable = true)\n",
        " |-- blockprefix: string (nullable = true)\n",
        " |-- blocksize: integer (nullable = true)\n",
        " |-- in_cnt: integer (nullable = true)\n",
        " |-- lock_time: integer (nullable = true)\n",
        " |-- magic_no: integer (nullable = true)\n",
        " |-- merklehash: string (nullable = true)\n",
        " |-- nonce: integer (nullable = true)\n",
        " |-- on_chain: integer (nullable = true)\n",
        " |-- out_cnt: integer (nullable = true)\n",
        " |-- prevhash: string (nullable = true)\n",
        " |-- time: integer (nullable = true)\n",
        " |-- transaction_cnt: integer (nullable = true)\n",
        " |-- txIn_prevhash: array (nullable = true)\n",
        " |    |-- element: string (containsNull = true)\n",
        " |-- txIn_prevtx_out_idx: array (nullable = true)\n",
        " |    |-- element: integer (containsNull = true)\n",
        " |-- txIn_scriptSig: array (nullable = true)\n",
        " |    |-- element: string (containsNull = true)\n",
        " |-- txIn_sequence_no: array (nullable = true)\n",
        " |    |-- element: integer (containsNull = true)\n",
        " |-- txIn_txin_script_len: array (nullable = true)\n",
        " |    |-- element: integer (containsNull = true)\n",
        " |-- txOut_scriptPubKey: array (nullable = true)\n",
        " |    |-- element: string (containsNull = true)\n",
        " |-- txOut_script_len: array (nullable = true)\n",
        " |    |-- element: integer (containsNull = true)\n",
        " |-- txOut_value: array (nullable = true)\n",
        " |    |-- element: integer (containsNull = true)\n",
        " |-- tx_version: integer (nullable = true)\n",
        " |-- version: integer (nullable = true)\n",
        "\n"
       ]
      }
     ],
     "prompt_number": 73
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "schTxRDD.saveAsPickleFile('schTxRDD.txt')"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "ename": "Py4JJavaError",
       "evalue": "An error occurred while calling o197.saveAsObjectFile.\n: org.apache.spark.SparkException: Job aborted due to stage failure: Task 2 in stage 24.0 failed 1 times, most recent failure: Lost task 2.0 in stage 24.0 (TID 35, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):\n  File \"/home/jeff/spark-1.2.1-bin-hadoop2.4/python/pyspark/worker.py\", line 92, in main\n    command = pickleSer.loads(command.value)\n  File \"/home/jeff/spark-1.2.1-bin-hadoop2.4/python/pyspark/broadcast.py\", line 106, in value\n    self._value = self.load(self._path)\n  File \"/home/jeff/spark-1.2.1-bin-hadoop2.4/python/pyspark/broadcast.py\", line 87, in load\n    with open(path, 'rb', 1 << 20) as f:\nIOError: [Errno 2] No such file or directory: '/home/jeff/spark-scratch/spark-13d28025-ac44-475f-a4cf-c265b55e6d19/spark-22fa63f0-2823-48bf-bfe4-28f7e5de418c/tmpzxB1wy'\n\n\tat org.apache.spark.api.python.PythonRDD$$anon$1.read(PythonRDD.scala:137)\n\tat org.apache.spark.api.python.PythonRDD$$anon$1.<init>(PythonRDD.scala:174)\n\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:96)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:280)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:247)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:280)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:247)\n\tat org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:280)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:247)\n\tat org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:280)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:247)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.sql.SchemaRDD.compute(SchemaRDD.scala:120)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:280)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:247)\n\tat org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:280)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:247)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:280)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:247)\n\tat org.apache.spark.api.python.PythonRDD$WriterThread$$anonfun$run$1.apply$mcV$sp(PythonRDD.scala:242)\n\tat org.apache.spark.api.python.PythonRDD$WriterThread$$anonfun$run$1.apply(PythonRDD.scala:204)\n\tat org.apache.spark.api.python.PythonRDD$WriterThread$$anonfun$run$1.apply(PythonRDD.scala:204)\n\tat org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:1550)\n\tat org.apache.spark.api.python.PythonRDD$WriterThread.run(PythonRDD.scala:203)\n\nDriver stacktrace:\n\tat org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1214)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1203)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1202)\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:1202)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:696)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:696)\n\tat scala.Option.foreach(Option.scala:236)\n\tat org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:696)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessActor$$anonfun$receive$2.applyOrElse(DAGScheduler.scala:1420)\n\tat akka.actor.Actor$class.aroundReceive(Actor.scala:465)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessActor.aroundReceive(DAGScheduler.scala:1375)\n\tat akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)\n\tat akka.actor.ActorCell.invoke(ActorCell.scala:487)\n\tat akka.dispatch.Mailbox.processMailbox(Mailbox.scala:238)\n\tat akka.dispatch.Mailbox.run(Mailbox.scala:220)\n\tat akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:393)\n\tat scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)\n\tat scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)\n\tat scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)\n\tat scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)\n",
       "output_type": "pyerr",
       "traceback": [
        "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m\n\u001b[1;31mPy4JJavaError\u001b[0m                             Traceback (most recent call last)",
        "\u001b[1;32m<ipython-input-48-a4534ba96a14>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m\n\u001b[1;32m----> 1\u001b[1;33m \u001b[0mschTxRDD\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msaveAsPickleFile\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m'schTxRDD.txt'\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m",
        "\u001b[1;32m/home/jeff/spark-1.2.1-bin-hadoop2.4/python/pyspark/rdd.py\u001b[0m in \u001b[0;36msaveAsPickleFile\u001b[1;34m(self, path, batchSize)\u001b[0m\n\u001b[0;32m   1265\u001b[0m         \u001b[1;32melse\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1266\u001b[0m             \u001b[0mser\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mBatchedSerializer\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mPickleSerializer\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mbatchSize\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m-> 1267\u001b[1;33m         \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_reserialize\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mser\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jrdd\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msaveAsObjectFile\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mpath\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m   1268\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1269\u001b[0m     \u001b[1;32mdef\u001b[0m \u001b[0msaveAsTextFile\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mpath\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
        "\u001b[1;32m/home/jeff/spark-1.2.1-bin-hadoop2.4/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    536\u001b[0m         \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[0;32m    537\u001b[0m         return_value = get_return_value(answer, self.gateway_client,\n\u001b[1;32m--> 538\u001b[1;33m                 self.target_id, self.name)\n\u001b[0m\u001b[0;32m    539\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    540\u001b[0m         \u001b[1;32mfor\u001b[0m \u001b[0mtemp_arg\u001b[0m \u001b[1;32min\u001b[0m \u001b[0mtemp_args\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
        "\u001b[1;32m/home/jeff/spark-1.2.1-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py\u001b[0m in \u001b[0;36mget_return_value\u001b[1;34m(answer, gateway_client, target_id, name)\u001b[0m\n\u001b[0;32m    298\u001b[0m                 raise Py4JJavaError(\n\u001b[0;32m    299\u001b[0m                     \u001b[1;34m'An error occurred while calling {0}{1}{2}.\\n'\u001b[0m\u001b[1;33m.\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 300\u001b[1;33m                     format(target_id, '.', name), value)\n\u001b[0m\u001b[0;32m    301\u001b[0m             \u001b[1;32melse\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    302\u001b[0m                 raise Py4JError(\n",
        "\u001b[1;31mPy4JJavaError\u001b[0m: An error occurred while calling o197.saveAsObjectFile.\n: org.apache.spark.SparkException: Job aborted due to stage failure: Task 2 in stage 24.0 failed 1 times, most recent failure: Lost task 2.0 in stage 24.0 (TID 35, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):\n  File \"/home/jeff/spark-1.2.1-bin-hadoop2.4/python/pyspark/worker.py\", line 92, in main\n    command = pickleSer.loads(command.value)\n  File \"/home/jeff/spark-1.2.1-bin-hadoop2.4/python/pyspark/broadcast.py\", line 106, in value\n    self._value = self.load(self._path)\n  File \"/home/jeff/spark-1.2.1-bin-hadoop2.4/python/pyspark/broadcast.py\", line 87, in load\n    with open(path, 'rb', 1 << 20) as f:\nIOError: [Errno 2] No such file or directory: '/home/jeff/spark-scratch/spark-13d28025-ac44-475f-a4cf-c265b55e6d19/spark-22fa63f0-2823-48bf-bfe4-28f7e5de418c/tmpzxB1wy'\n\n\tat org.apache.spark.api.python.PythonRDD$$anon$1.read(PythonRDD.scala:137)\n\tat org.apache.spark.api.python.PythonRDD$$anon$1.<init>(PythonRDD.scala:174)\n\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:96)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:280)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:247)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:280)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:247)\n\tat org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:280)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:247)\n\tat org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:280)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:247)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.sql.SchemaRDD.compute(SchemaRDD.scala:120)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:280)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:247)\n\tat org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:280)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:247)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:280)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:247)\n\tat org.apache.spark.api.python.PythonRDD$WriterThread$$anonfun$run$1.apply$mcV$sp(PythonRDD.scala:242)\n\tat org.apache.spark.api.python.PythonRDD$WriterThread$$anonfun$run$1.apply(PythonRDD.scala:204)\n\tat org.apache.spark.api.python.PythonRDD$WriterThread$$anonfun$run$1.apply(PythonRDD.scala:204)\n\tat org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:1550)\n\tat org.apache.spark.api.python.PythonRDD$WriterThread.run(PythonRDD.scala:203)\n\nDriver stacktrace:\n\tat org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1214)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1203)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1202)\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:1202)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:696)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:696)\n\tat scala.Option.foreach(Option.scala:236)\n\tat org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:696)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessActor$$anonfun$receive$2.applyOrElse(DAGScheduler.scala:1420)\n\tat akka.actor.Actor$class.aroundReceive(Actor.scala:465)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessActor.aroundReceive(DAGScheduler.scala:1375)\n\tat akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)\n\tat akka.actor.ActorCell.invoke(ActorCell.scala:487)\n\tat akka.dispatch.Mailbox.processMailbox(Mailbox.scala:238)\n\tat akka.dispatch.Mailbox.run(Mailbox.scala:220)\n\tat akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:393)\n\tat scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)\n\tat scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)\n\tat scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)\n\tat scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)\n"
       ]
      }
     ],
     "prompt_number": 48
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "%%bash\n",
      "rm -r /home/jeff/prj/bitcoin-analytics/temp.parquet"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 73
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "schTxRDD.saveAsParquetFile('temp.parquet')"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "ename": "Py4JJavaError",
       "evalue": "An error occurred while calling o273.saveAsParquetFile.\n: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 47.0 failed 1 times, most recent failure: Lost task 0.0 in stage 47.0 (TID 69, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):\n  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/worker.py\", line 92, in main\n    if isinstance(command, Broadcast):\n  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/broadcast.py\", line 112, in value\n    self._value = self.load(self._path)\n  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/broadcast.py\", line 92, in load\n    with open(path, 'rb', 1 << 20) as f:\nIOError: [Errno 2] No such file or directory: '/home/jeff/spark-scratch/spark-b1eee178-7bfd-4442-961b-80055c4aab79/tmpgffDkv'\n\n\tat org.apache.spark.api.python.PythonRDD$$anon$1.read(PythonRDD.scala:137)\n\tat org.apache.spark.api.python.PythonRDD$$anon$1.<init>(PythonRDD.scala:174)\n\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:96)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61)\n\tat org.apache.spark.scheduler.Task.run(Task.scala:56)\n\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196)\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:745)\n\nDriver stacktrace:\n\tat org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1214)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1203)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1202)\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:1202)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:696)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:696)\n\tat scala.Option.foreach(Option.scala:236)\n\tat org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:696)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessActor$$anonfun$receive$2.applyOrElse(DAGScheduler.scala:1420)\n\tat akka.actor.Actor$class.aroundReceive(Actor.scala:465)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessActor.aroundReceive(DAGScheduler.scala:1375)\n\tat akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)\n\tat akka.actor.ActorCell.invoke(ActorCell.scala:487)\n\tat akka.dispatch.Mailbox.processMailbox(Mailbox.scala:238)\n\tat akka.dispatch.Mailbox.run(Mailbox.scala:220)\n\tat akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:393)\n\tat scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)\n\tat scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)\n\tat scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)\n\tat scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)\n",
       "output_type": "pyerr",
       "traceback": [
        "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m\n\u001b[1;31mPy4JJavaError\u001b[0m                             Traceback (most recent call last)",
        "\u001b[1;32m<ipython-input-74-bd3cff4ebbed>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m\n\u001b[1;32m----> 1\u001b[1;33m \u001b[0mschTxRDD\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msaveAsParquetFile\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m'temp.parquet'\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m",
        "\u001b[1;32m/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/sql.pyc\u001b[0m in \u001b[0;36msaveAsParquetFile\u001b[1;34m(self, path)\u001b[0m\n\u001b[0;32m   1901\u001b[0m         \u001b[0mTrue\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1902\u001b[0m         \"\"\"\n\u001b[1;32m-> 1903\u001b[1;33m         \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jschema_rdd\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msaveAsParquetFile\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mpath\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m   1904\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1905\u001b[0m     \u001b[1;32mdef\u001b[0m \u001b[0mregisterTempTable\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mname\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
        "\u001b[1;32m/home/jeff/spark-1.2.0-bin-hadoop2.4/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    536\u001b[0m         \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[0;32m    537\u001b[0m         return_value = get_return_value(answer, self.gateway_client,\n\u001b[1;32m--> 538\u001b[1;33m                 self.target_id, self.name)\n\u001b[0m\u001b[0;32m    539\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    540\u001b[0m         \u001b[1;32mfor\u001b[0m \u001b[0mtemp_arg\u001b[0m \u001b[1;32min\u001b[0m \u001b[0mtemp_args\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
        "\u001b[1;32m/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py\u001b[0m in \u001b[0;36mget_return_value\u001b[1;34m(answer, gateway_client, target_id, name)\u001b[0m\n\u001b[0;32m    298\u001b[0m                 raise Py4JJavaError(\n\u001b[0;32m    299\u001b[0m                     \u001b[1;34m'An error occurred while calling {0}{1}{2}.\\n'\u001b[0m\u001b[1;33m.\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 300\u001b[1;33m                     format(target_id, '.', name), value)\n\u001b[0m\u001b[0;32m    301\u001b[0m             \u001b[1;32melse\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    302\u001b[0m                 raise Py4JError(\n",
        "\u001b[1;31mPy4JJavaError\u001b[0m: An error occurred while calling o273.saveAsParquetFile.\n: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 47.0 failed 1 times, most recent failure: Lost task 0.0 in stage 47.0 (TID 69, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):\n  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/worker.py\", line 92, in main\n    if isinstance(command, Broadcast):\n  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/broadcast.py\", line 112, in value\n    self._value = self.load(self._path)\n  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/broadcast.py\", line 92, in load\n    with open(path, 'rb', 1 << 20) as f:\nIOError: [Errno 2] No such file or directory: '/home/jeff/spark-scratch/spark-b1eee178-7bfd-4442-961b-80055c4aab79/tmpgffDkv'\n\n\tat org.apache.spark.api.python.PythonRDD$$anon$1.read(PythonRDD.scala:137)\n\tat org.apache.spark.api.python.PythonRDD$$anon$1.<init>(PythonRDD.scala:174)\n\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:96)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61)\n\tat org.apache.spark.scheduler.Task.run(Task.scala:56)\n\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196)\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:745)\n\nDriver stacktrace:\n\tat org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1214)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1203)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1202)\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:1202)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:696)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:696)\n\tat scala.Option.foreach(Option.scala:236)\n\tat org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:696)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessActor$$anonfun$receive$2.applyOrElse(DAGScheduler.scala:1420)\n\tat akka.actor.Actor$class.aroundReceive(Actor.scala:465)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessActor.aroundReceive(DAGScheduler.scala:1375)\n\tat akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)\n\tat akka.actor.ActorCell.invoke(ActorCell.scala:487)\n\tat akka.dispatch.Mailbox.processMailbox(Mailbox.scala:238)\n\tat akka.dispatch.Mailbox.run(Mailbox.scala:220)\n\tat akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:393)\n\tat scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)\n\tat scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)\n\tat scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)\n\tat scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)\n"
       ]
      }
     ],
     "prompt_number": 74
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "print(res.take(1))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "ename": "Py4JJavaError",
       "evalue": "An error occurred while calling o183.collectToPython.\n: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 24.0 failed 1 times, most recent failure: Lost task 0.0 in stage 24.0 (TID 45, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):\n  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/worker.py\", line 92, in main\n    command = pickleSer.loads(command.value)\n  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/broadcast.py\", line 113, in value\n    if self._jbroadcast is None:\n  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/broadcast.py\", line 94, in load\n    try:\nIOError: [Errno 2] No such file or directory: '/home/jeff/spark-scratch/spark-4c103b7f-93fc-473e-8a0b-7d629f325f53/tmpUWgmUY'\n\n\tat org.apache.spark.api.python.PythonRDD$$anon$1.read(PythonRDD.scala:137)\n\tat org.apache.spark.api.python.PythonRDD$$anon$1.<init>(PythonRDD.scala:174)\n\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:96)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61)\n\tat org.apache.spark.scheduler.Task.run(Task.scala:56)\n\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196)\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:745)\n\nDriver stacktrace:\n\tat org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1214)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1203)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1202)\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:1202)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:696)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:696)\n\tat scala.Option.foreach(Option.scala:236)\n\tat org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:696)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessActor$$anonfun$receive$2.applyOrElse(DAGScheduler.scala:1420)\n\tat akka.actor.Actor$class.aroundReceive(Actor.scala:465)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessActor.aroundReceive(DAGScheduler.scala:1375)\n\tat akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)\n\tat akka.actor.ActorCell.invoke(ActorCell.scala:487)\n\tat akka.dispatch.Mailbox.processMailbox(Mailbox.scala:238)\n\tat akka.dispatch.Mailbox.run(Mailbox.scala:220)\n\tat akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:393)\n\tat scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)\n\tat scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)\n\tat scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)\n\tat scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)\n",
       "output_type": "pyerr",
       "traceback": [
        "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m\n\u001b[1;31mPy4JJavaError\u001b[0m                             Traceback (most recent call last)",
        "\u001b[1;32m<ipython-input-47-a4d1e36b34f0>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m\n\u001b[1;32m----> 1\u001b[1;33m \u001b[1;32mprint\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mres\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mtake\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[1;32m/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/sql.pyc\u001b[0m in \u001b[0;36mtake\u001b[1;34m(self, num)\u001b[0m\n\u001b[0;32m   1994\u001b[0m         \u001b[1;33m[\u001b[0m\u001b[0mRow\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mfield1\u001b[0m\u001b[1;33m=\u001b[0m\u001b[1;36m1\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mfield2\u001b[0m\u001b[1;33m=\u001b[0m\u001b[1;34mu'row1'\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mRow\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mfield1\u001b[0m\u001b[1;33m=\u001b[0m\u001b[1;36m2\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mfield2\u001b[0m\u001b[1;33m=\u001b[0m\u001b[1;34mu'row2'\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m]\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1995\u001b[0m         \"\"\"\n\u001b[1;32m-> 1996\u001b[1;33m         \u001b[1;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mlimit\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mnum\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mcollect\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   1997\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1998\u001b[0m     \u001b[1;31m# Convert each object in the RDD to a Row with the right class\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
        "\u001b[1;32m/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/sql.pyc\u001b[0m in \u001b[0;36mcollect\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m   1976\u001b[0m         \"\"\"\n\u001b[0;32m   1977\u001b[0m         \u001b[1;32mwith\u001b[0m \u001b[0mSCCallSiteSync\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mcontext\u001b[0m\u001b[1;33m)\u001b[0m \u001b[1;32mas\u001b[0m \u001b[0mcss\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m-> 1978\u001b[1;33m             \u001b[0mbytesInJava\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jschema_rdd\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mbaseSchemaRDD\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mcollectToPython\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0miterator\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   1979\u001b[0m         \u001b[0mcls\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0m_create_cls\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mschema\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1980\u001b[0m         \u001b[1;32mreturn\u001b[0m \u001b[0mmap\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mcls\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_collect_iterator_through_file\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mbytesInJava\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
        "\u001b[1;32m/home/jeff/spark-1.2.0-bin-hadoop2.4/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    536\u001b[0m         \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[0;32m    537\u001b[0m         return_value = get_return_value(answer, self.gateway_client,\n\u001b[1;32m--> 538\u001b[1;33m                 self.target_id, self.name)\n\u001b[0m\u001b[0;32m    539\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    540\u001b[0m         \u001b[1;32mfor\u001b[0m \u001b[0mtemp_arg\u001b[0m \u001b[1;32min\u001b[0m \u001b[0mtemp_args\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
        "\u001b[1;32m/home/jeff/spark-1.2.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py\u001b[0m in \u001b[0;36mget_return_value\u001b[1;34m(answer, gateway_client, target_id, name)\u001b[0m\n\u001b[0;32m    298\u001b[0m                 raise Py4JJavaError(\n\u001b[0;32m    299\u001b[0m                     \u001b[1;34m'An error occurred while calling {0}{1}{2}.\\n'\u001b[0m\u001b[1;33m.\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 300\u001b[1;33m                     format(target_id, '.', name), value)\n\u001b[0m\u001b[0;32m    301\u001b[0m             \u001b[1;32melse\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    302\u001b[0m                 raise Py4JError(\n",
        "\u001b[1;31mPy4JJavaError\u001b[0m: An error occurred while calling o183.collectToPython.\n: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 24.0 failed 1 times, most recent failure: Lost task 0.0 in stage 24.0 (TID 45, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):\n  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/worker.py\", line 92, in main\n    command = pickleSer.loads(command.value)\n  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/broadcast.py\", line 113, in value\n    if self._jbroadcast is None:\n  File \"/home/jeff/spark-1.2.0-bin-hadoop2.4/python/pyspark/broadcast.py\", line 94, in load\n    try:\nIOError: [Errno 2] No such file or directory: '/home/jeff/spark-scratch/spark-4c103b7f-93fc-473e-8a0b-7d629f325f53/tmpUWgmUY'\n\n\tat org.apache.spark.api.python.PythonRDD$$anon$1.read(PythonRDD.scala:137)\n\tat org.apache.spark.api.python.PythonRDD$$anon$1.<init>(PythonRDD.scala:174)\n\tat org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:96)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31)\n\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)\n\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:230)\n\tat org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61)\n\tat org.apache.spark.scheduler.Task.run(Task.scala:56)\n\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196)\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:745)\n\nDriver stacktrace:\n\tat org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1214)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1203)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1202)\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:1202)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:696)\n\tat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:696)\n\tat scala.Option.foreach(Option.scala:236)\n\tat org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:696)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessActor$$anonfun$receive$2.applyOrElse(DAGScheduler.scala:1420)\n\tat akka.actor.Actor$class.aroundReceive(Actor.scala:465)\n\tat org.apache.spark.scheduler.DAGSchedulerEventProcessActor.aroundReceive(DAGScheduler.scala:1375)\n\tat akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)\n\tat akka.actor.ActorCell.invoke(ActorCell.scala:487)\n\tat akka.dispatch.Mailbox.processMailbox(Mailbox.scala:238)\n\tat akka.dispatch.Mailbox.run(Mailbox.scala:220)\n\tat akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:393)\n\tat scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)\n\tat scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)\n\tat scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)\n\tat scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)\n"
       ]
      }
     ],
     "prompt_number": 47
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "tmp = {}\n",
      "tmp['key1'] = []\n",
      "tmp['key2'] = 'cat'"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 74
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "print(tmp.get('key1',[]).append(10))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "None\n"
       ]
      }
     ],
     "prompt_number": 77
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "tmp.get('key3',[])"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 78,
       "text": [
        "[]"
       ]
      }
     ],
     "prompt_number": 78
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "print(tmp.get('key3',[]).append(10))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "None\n"
       ]
      }
     ],
     "prompt_number": 80
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "test = [].append(10)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 81
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "print(test)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "None\n"
       ]
      }
     ],
     "prompt_number": 83
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "tmp.get('key1').append(90)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 85
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "print(tmp)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "{'key2': 'cat', 'key1': [10, 10, 90]}\n"
       ]
      }
     ],
     "prompt_number": 86
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "tmp.get('key100',[]).append(80)\n",
      "print(tmp)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "{'key2': 'cat', 'key1': [10, 10, 90]}\n"
       ]
      }
     ],
     "prompt_number": 87
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "tmp.get(')"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    }
   ],
   "metadata": {}
  }
 ]
}