Schema_Evolution_Exploration.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The aim of this notebook is to express the semantics of schema evolution of parquet files using both Spark and parquet2hive + Presto. Note that all parquet2hive is doing is reading the schema *from the most recently created file*, so in some cases this could be changed without changing Presto's underlying facilities by reading *all* files.\n",
    "\n",
    "In this notebook I am running Spark locally, and using a remote Presto cluster. To connect to this cluster I'm using parquet2hive_server [0], which is just a simple API for parquet2hive on the remote cluster. To run this notebook successfully, you'll need to run the following on the Presto cluster:\n",
    "\n",
    "```\n",
    "sudo pip install parquet2hive_server\n",
    "start_parquet2hive_server\n",
    "```\n",
    "\n",
    "[0] http://www.github.com/fbertsch/parquet2hive_server"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from parquet2hive_server.client import Parquet2HiveClient\n",
    "from pyhive import presto\n",
    "from pprint import pprint\n",
    "\n",
    "presto_dns = 'ec2-54-149-100-125.us-west-2.compute.amazonaws.com'\n",
    "\n",
    "client = Parquet2HiveClient(presto_dns + ':5129')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import boto3\n",
    "\n",
    "bucket, prefix = \"telemetry-test-bucket\", \"schema_evolution\"\n",
    "dataset = \"s3://{}/{}/\".format(bucket, prefix)\n",
    "partition = '/type='\n",
    "\n",
    "s3 = boto3.resource('s3')\n",
    "objects_to_delete = s3.meta.client.list_objects(Bucket=bucket, Prefix=prefix)\n",
    "\n",
    "delete_keys = {}\n",
    "delete_keys['Objects'] = [{'Key' : k} for k in [obj['Key'] for obj in objects_to_delete.get('Contents', [])]]\n",
    "\n",
    "try:\n",
    "    _ = s3.meta.client.delete_objects(Bucket=bucket, Delete=delete_keys)\n",
    "except Exception:\n",
    "    pass"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "conn = presto.connect(host=presto_dns, port='8889')\n",
    "cursor = conn.cursor()\n",
    "\n",
    "def get_schema(_cursor, _v):\n",
    "    \"\"\"Prints the schema in a similar format to spark's dataframe.printSchema()\"\"\"\n",
    "    _cursor.execute('describe schema_evolution_{}'.format(_v))\n",
    "    return '\\\\\\n'.join(['root'] + [' |-- {}: {}'.format(a, b) for a, b, _, _ in _cursor.fetchall()])\n",
    "\n",
    "def execute(_cursor, _query):\n",
    "    _cursor.execute(_query)\n",
    "    results = _cursor.fetchall()\n",
    "    colnames = [c[0] for c in _cursor.description]\n",
    "    return '\\\\\\n'.join([', '.join(['{}={}'.format(c,r) for c,r in zip(colnames, res)]) for res in results])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "## Adding a Column"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "v = 'v1'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([(0,),(1,)], 1)\n",
    "df = sqlContext.createDataFrame(rdd, ['id'])\n",
    "df.write.parquet(dataset + v + partition + '1')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([(0,0),(1,1)], 1)\n",
    "df = sqlContext.createDataFrame(rdd, ['id', 'score'])\n",
    "df.write.parquet(dataset + v + partition  + '2')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.load(dataset + v, 'parquet').printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.load(dataset + v, 'parquet').collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.option(\"mergeSchema\", \"true\").load(dataset + v, 'parquet').printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.option(\"mergeSchema\", \"true\").load(dataset + v, 'parquet').collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Presto"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "client.load(dataset=dataset, dv=v)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print get_schema(cursor, v)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print execute(cursor, \"SELECT * FROM schema_evolution_{}\".format(v))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Removing a Column"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "v = 'v2'\n",
    "\n",
    "rdd = sc.parallelize([(0,0),(1,1)], 1)\n",
    "df = sqlContext.createDataFrame(rdd, ['id', 'score'])\n",
    "df.write.parquet(dataset + v + partition  + '1')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([(0,),(1,)], 1)\n",
    "df = sqlContext.createDataFrame(rdd, ['id'])\n",
    "df.write.parquet(dataset + v + partition + '2')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.load(dataset + v, 'parquet').printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.load(dataset + v, 'parquet').collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.option(\"mergeSchema\", \"true\").load(dataset + v, 'parquet').printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.option(\"mergeSchema\", \"true\").load(dataset + v, 'parquet').collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Presto"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "client.load(dataset=dataset, dv=v)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print get_schema(cursor, v)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print execute(cursor, \"SELECT * FROM schema_evolution_{}\".format(v))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Renaming a Column"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "v = 'v3'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([(0,),(1,)], 1)\n",
    "df = sqlContext.createDataFrame(rdd, ['id'])\n",
    "df.write.parquet(dataset + v + partition + '1')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([(0,),(1,)], 1)\n",
    "df = sqlContext.createDataFrame(rdd, ['score'])\n",
    "df.write.parquet(dataset + v + partition + '2')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.load(dataset + v, 'parquet').printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.load(dataset + v, 'parquet').collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.option(\"mergeSchema\", \"true\").load(dataset + v, 'parquet').printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.option(\"mergeSchema\", \"true\").load(dataset + v, 'parquet').collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Presto"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "client.load(dataset=dataset, dv=v)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print get_schema(cursor, v)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print execute(cursor, \"SELECT * FROM schema_evolution_{}\".format(v))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Replace Column"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note that this is similar to \"rename column\", but the new data has a different type."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "v = 'v4'\n",
    "\n",
    "rdd = sc.parallelize([(0,),(1,)], 1)\n",
    "df = sqlContext.createDataFrame(rdd, ['id'])\n",
    "df.write.parquet(dataset + v + partition + '1')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([('a',),('b',)], 1)\n",
    "df = sqlContext.createDataFrame(rdd, ['score'])\n",
    "df.write.parquet(dataset + v + partition + '2')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.load(dataset + v, 'parquet').printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df = sqlContext.read.load(dataset + v, 'parquet').collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.option(\"mergeSchema\", \"true\").load(dataset + v, 'parquet').printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.option(\"mergeSchema\", \"true\").load(dataset + v, 'parquet').collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Presto"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "client.load(dataset=dataset, dv=v)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print get_schema(cursor, v)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print execute(cursor, \"SELECT * FROM schema_evolution_{}\".format(v))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Transpose Columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "v = 'v5'\n",
    "\n",
    "rdd = sc.parallelize([(0,'a','b')], 1)\n",
    "df = sqlContext.createDataFrame(rdd, ['id', 'transpose_a', 'transpose_b'])\n",
    "df.write.parquet(dataset + v + partition + '1')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([(1,'b','a')], 1)\n",
    "df = sqlContext.createDataFrame(rdd, ['id', 'transpose_b', 'transpose_a'])\n",
    "df.write.parquet(dataset + v + partition + '2')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "### Spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.load(dataset + v, 'parquet').printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.load(dataset + v, 'parquet').select('transpose_a','transpose_b').collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "sqlContext.read.option(\"mergeSchema\", \"true\").load(dataset + v, 'parquet').printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.option(\"mergeSchema\", \"true\").load(dataset + v, 'parquet').select('transpose_a','transpose_b').collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Presto"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "client.load(dataset=dataset, dv=v)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print(get_schema(cursor, v))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print execute(cursor, \"SELECT * FROM schema_evolution_{}\".format(v))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Transpose, Delete and Add Columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "v = 'v6'\n",
    "\n",
    "rdd = sc.parallelize([(0,'r','t')], 1)\n",
    "df = sqlContext.createDataFrame(rdd, ['id', 'removed', 'transposed'])\n",
    "df.write.parquet(dataset + v + partition + '1')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([(1,'t','a')], 1)\n",
    "df = sqlContext.createDataFrame(rdd, ['id', 'transposed', 'added'])\n",
    "df.write.parquet(dataset + v + partition + '2')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.load(dataset + v, 'parquet').printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.load(dataset + v, 'parquet').collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.option(\"mergeSchema\", \"true\").load(dataset + v, 'parquet').printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.option(\"mergeSchema\", \"true\").load(dataset + v, 'parquet').collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Presto"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "client.load(dataset=dataset, dv=v)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print(get_schema(cursor, v))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print execute(cursor, \"SELECT * FROM schema_evolution_{}\".format(v))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Nested Row Type - Adding a Subcolumn"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark.sql.types import *\n",
    "\n",
    "v = 'v7'\n",
    "\n",
    "df = sqlContext.createDataFrame([[[1, 'e']]], StructType([\n",
    "    StructField(\"nested\", \n",
    "                StructType([\n",
    "                        StructField(\"id\", LongType()), \n",
    "                        StructField(\"exists\", StringType())\n",
    "                    ])\n",
    "               )\n",
    "]))\n",
    "\n",
    "df.write.parquet(dataset + v + partition + '1')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "df = sqlContext.createDataFrame([[[1, 'e', 'a']]], StructType([\n",
    "    StructField(\"nested\", \n",
    "                StructType([\n",
    "                        StructField(\"id\", LongType()), \n",
    "                        StructField(\"exists\", StringType()),\n",
    "                        StructField(\"added\", StringType())\n",
    "                    ])\n",
    "               )\n",
    "]))\n",
    "df.write.parquet(dataset + v + partition + '2')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.load(dataset + v, 'parquet').printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.load(dataset + v, 'parquet').collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.option(\"mergeSchema\", \"true\").load(dataset + v, 'parquet').printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.option(\"mergeSchema\", \"true\").load(dataset + v, 'parquet').collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Presto"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "client.load(dataset=dataset, dv=v)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print(get_schema(cursor, v))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print execute(cursor, \"SELECT * FROM schema_evolution_{}\".format(v))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Nested Row Type - Removing a Subcolumn"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "v = 'v8'\n",
    "\n",
    "df = sqlContext.createDataFrame([[[1, 'e', 'r']]], StructType([\n",
    "    StructField(\"nested\", \n",
    "                StructType([\n",
    "                        StructField(\"id\", LongType()), \n",
    "                        StructField(\"exists\", StringType()),\n",
    "                        StructField(\"removed\", StringType())\n",
    "                    ])\n",
    "               )\n",
    "]))\n",
    "\n",
    "df.write.parquet(dataset + v + partition + '1')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "df = sqlContext.createDataFrame([[[1, 'e']]], StructType([\n",
    "    StructField(\"nested\", \n",
    "                StructType([\n",
    "                        StructField(\"id\", LongType()), \n",
    "                        StructField(\"exists\", StringType())\n",
    "                    ])\n",
    "               )\n",
    "]))\n",
    "\n",
    "df.write.parquet(dataset + v + partition + '2')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.load(dataset + v, 'parquet').printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.load(dataset + v, 'parquet').collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.option(\"mergeSchema\", \"true\").load(dataset + v, 'parquet').printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sqlContext.read.option(\"mergeSchema\", \"true\").load(dataset + v, 'parquet').collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Presto"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "client.load(dataset=dataset, dv=v)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print(get_schema(cursor, v))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print execute(cursor, \"SELECT * FROM schema_evolution_{}\".format(v))"
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "Python [default]",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}