Module 2 - PySpark DataFrames.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "people = spark.read.json('resources/people.json')\n",
    "df = spark.read.json('resources/people.json')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "users = spark.read.parquet('resources/users.parquet')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "employees = spark.read.json('resources/employees.json')\n",
    "df2 = spark.read.json('resources/employees.json')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+\n",
      "| age|   name|\n",
      "+----+-------+\n",
      "|null|Michael|\n",
      "|  30|   Andy|\n",
      "|  19| Justin|\n",
      "+----+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "people.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+--------------+----------------+\n",
      "|  name|favorite_color|favorite_numbers|\n",
      "+------+--------------+----------------+\n",
      "|Alyssa|          null|  [3, 9, 15, 20]|\n",
      "|   Ben|           red|              []|\n",
      "+------+--------------+----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "users.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------+\n",
      "|   name|salary|\n",
      "+-------+------+\n",
      "|Michael|  3000|\n",
      "|   Andy|  4500|\n",
      "| Justin|  3500|\n",
      "|  Berta|  4000|\n",
      "+-------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "employees.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### agg\n",
    "* Aggregate on the entire DataFrame without groups (shorthand for df.groupBy.agg())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(max(age)=30)]"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.agg({\"age\": \"max\"}).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(min(age)=19)]"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.sql import functions as F\n",
    "df.agg(F.min(df.age)).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### alias\n",
    "* Returns a new DataFrame with an alias set."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(name=u'Michael', name=u'Michael', age=None),\n",
       " Row(name=u'Andy', name=u'Andy', age=30),\n",
       " Row(name=u'Justin', name=u'Justin', age=19)]"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.sql.functions import *\n",
    "df_as1 = df.alias(\"df_as1\")\n",
    "df_as2 = df.alias(\"df_as2\")\n",
    "joined_df = df_as1.join(df_as2, col(\"df_as1.name\") == col(\"df_as2.name\"), 'inner')\n",
    "joined_df.select(\"df_as1.name\", \"df_as2.name\", \"df_as2.age\").collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Works the same way as RDD\n",
    "#### cache\n",
    "#### checkpoint\n",
    "#### coalesce\n",
    "#### collect\n",
    "* Returns rows\n",
    "\n",
    "#### count\n",
    "* Returns num of rows\n",
    "<hr>\n",
    "<hr>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### columns\n",
    "* Returns all column names as list"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['age', 'name']"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### corr\n",
    "* Calculates the correlation of two columns of a DataFrame as a double value. \n",
    "* Currently only supports the Pearson Correlation Coefficient."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1.0"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "cdf = spark.createDataFrame([(1, 4), (2, 5), (3, 6)], [\"A\", \"B\"])\n",
    "\n",
    "cdf.corr('A','B')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### createGlobalTempView\n",
    "* Creates a global temporary view with this DataFrame.\n",
    "* The lifetime of this temporary view is tied to this <u>Spark application</u>\n",
    "* throws TempTableAlreadyExistsException, if the view name already exists in the catalog."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "df.createGlobalTempView(\"people\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "df2 = spark.sql(\"select * from global_temp.people\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sorted(df.collect()) == sorted(df2.collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(age=None, name=u'Michael'),\n",
       " Row(age=30, name=u'Andy'),\n",
       " Row(age=19, name=u'Justin')]"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df2.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "spark.catalog.dropGlobalTempView(\"people\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### createOrReplaceGlobalTempView\n",
    "* Creates or replaces a global temporary view using the given name.\n",
    "* The lifetime of this temporary view is tied to this <u>Spark application.</u>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### createTempView\n",
    "* Creates a local temporary view with this DataFrame.\n",
    "* The lifetime of this temporary table is tied to the SparkSession that was used to create this DataFrame. \n",
    "* throws TempTableAlreadyExistsException, if the view name already exists in the catalog."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.createTempView(\"people\")\n",
    "df2 = spark.sql(\"select * from people\")\n",
    "sorted(df.collect()) == sorted(df2.collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### createOrReplaceTempView\n",
    "* Creates or replaces a local temporary view with this DataFrame.\n",
    "* The lifetime of this temporary table is tied to the SparkSession that was used to create this DataFrame."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### describe\n",
    "* Computes statistics for numeric and string columns.\n",
    "* This include count, mean, stddev, min, and max. \n",
    "* If no columns are given, this function computes statistics for all numerical or string columns."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------------------+-------+\n",
      "|summary|               age|   name|\n",
      "+-------+------------------+-------+\n",
      "|  count|                 2|      3|\n",
      "|   mean|              24.5|   null|\n",
      "| stddev|7.7781745930520225|   null|\n",
      "|    min|                19|   Andy|\n",
      "|    max|                30|Michael|\n",
      "+-------+------------------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.describe().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### distinct\n",
    "* Returns a new DataFrame containing the distinct rows in this DataFrame."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(age=None, name=u'Michael'),\n",
       " Row(age=30, name=u'Andy'),\n",
       " Row(age=19, name=u'Justin')]"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.distinct().collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### drop\n",
    "* Returns a new DataFrame that drops the specified column"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(name=u'Michael'), Row(name=u'Andy'), Row(name=u'Justin')]"
      ]
     },
     "execution_count": 38,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df = spark.read.json('resources/people.json')\n",
    "df.drop('age').collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(name=u'Michael'), Row(name=u'Andy'), Row(name=u'Justin')]"
      ]
     },
     "execution_count": 39,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df = spark.read.json('resources/people.json')\n",
    "df.drop(df.age).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(age=None, name=u'Michael', salary=3000),\n",
       " Row(age=30, name=u'Andy', salary=4500),\n",
       " Row(age=19, name=u'Justin', salary=3500)]"
      ]
     },
     "execution_count": 40,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.join(df2, df.name == df2.name, 'inner').drop(df.name).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(name=u'Michael', salary=3000),\n",
       " Row(name=u'Andy', salary=4500),\n",
       " Row(name=u'Justin', salary=3500)]"
      ]
     },
     "execution_count": 41,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.join(df2, 'name', 'inner').drop('age', 'height').collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### dropduplicates\n",
    "* Return a new DataFrame with duplicate rows removed, optionally only considering certain columns.\n",
    "* For a static batch DataFrame, it just drops duplicate rows."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import Row\n",
    "f = sc.parallelize([ \\\n",
    "     Row(name='Alice', age=5, height=80), \\\n",
    "     Row(name='Alice', age=5, height=80), \\\n",
    "     Row(name='Alice', age=10, height=80)]).toDF()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+------+-----+\n",
      "|age|height| name|\n",
      "+---+------+-----+\n",
      "|  5|    80|Alice|\n",
      "| 10|    80|Alice|\n",
      "+---+------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "f.dropDuplicates().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+------+-----+\n",
      "|age|height| name|\n",
      "+---+------+-----+\n",
      "|  5|    80|Alice|\n",
      "+---+------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "f.dropDuplicates(['name', 'height']).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### dropna\n",
    "* Returns a new DataFrame omitting rows with null values\n",
    "* how – ‘any’ or ‘all’. If ‘any’, drop a row if it contains any nulls. If ‘all’, drop a row only if all its values are null.\n",
    "* thresh – int, default None If specified, drop rows that have less than thresh non-null values. This overwrites the how parameter.\n",
    "* subset – optional list of column names to consider."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+------+\n",
      "|age|  name|\n",
      "+---+------+\n",
      "| 30|  Andy|\n",
      "| 19|Justin|\n",
      "+---+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.na.drop(how='any').show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### explain\n",
    "* Prints the (logical and physical) plans to the console for debugging purpose."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "== Physical Plan ==\n",
      "*FileScan json [age#524L,name#525] Batched: false, Format: JSON, Location: InMemoryFileIndex[file:/tensorFolder/pyspark/resources/people.json], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<age:bigint,name:string>\n"
     ]
    }
   ],
   "source": [
    "df.explain()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### fillna\n",
    "* Replace null values"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-------+\n",
      "|age|   name|\n",
      "+---+-------+\n",
      "| 50| joseph|\n",
      "| 12|  james|\n",
      "| 31|unknown|\n",
      "+---+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df4 = spark.createDataFrame([(None, \"joseph\"), (12, \"james\"), (31, None)], [\"age\", \"name\"])\n",
    "df4.na.fill({'age': 50, 'name': 'unknown'}).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### filter\n",
    "* Filters rows using the given condition.\n",
    "* where is alias"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(age=30, name=u'Andy')]"
      ]
     },
     "execution_count": 53,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.filter(df.age > 20).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(age=19, name=u'Justin')]"
      ]
     },
     "execution_count": 54,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.where(df.age == 19).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(age=19, name=u'Justin')]"
      ]
     },
     "execution_count": 55,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.where(\"age = 19\").collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Same as RDD\n",
    "#### first\n",
    "#### foreach\n",
    "#### foreachPartition"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### groupBy\n",
    "* Groups the DataFrame using the specified columns, so we can run aggregation on them. \n",
    "* See GroupedData for all the available aggregate functions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(avg(age)=24.5)]"
      ]
     },
     "execution_count": 57,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.groupBy().avg().collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(name=u'Michael', avg(age)=None),\n",
       " Row(name=u'Andy', avg(age)=30.0),\n",
       " Row(name=u'Justin', avg(age)=19.0)]"
      ]
     },
     "execution_count": 58,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.groupBy('name').agg({'age': 'mean'}).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(name=u'Michael', avg(age)=None),\n",
       " Row(name=u'Andy', avg(age)=30.0),\n",
       " Row(name=u'Justin', avg(age)=19.0)]"
      ]
     },
     "execution_count": 59,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.groupBy(df.name).avg().collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(name=u'Andy', age=30, count=1),\n",
       " Row(name=u'Michael', age=None, count=1),\n",
       " Row(name=u'Justin', age=19, count=1)]"
      ]
     },
     "execution_count": 60,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.groupBy(['name', df.age]).count().collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### head\n",
    "* First n rows"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(age=None, name=u'Michael')"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### intersect\n",
    "* Returns dataframe with common rows"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#### join"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(name=u'Michael'), Row(name=u'Andy'), Row(name=None), Row(name=u'Justin')]"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.join(df2, df.name == df2.name, 'outer').select(df.name).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#### schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "StructType(List(StructField(age,LongType,true),StructField(name,StringType,true)))"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#### select"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(age=None, name=u'Michael'),\n",
       " Row(age=30, name=u'Andy'),\n",
       " Row(age=19, name=u'Justin')]"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.select('*').collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(name=u'Michael', age=None),\n",
       " Row(name=u'Andy', age=40),\n",
       " Row(name=u'Justin', age=29)]"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.select(df.name, (df.age + 10).alias('age')).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#### selectExpr"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row((age * 2)=None, abs(age)=None),\n",
       " Row((age * 2)=60, abs(age)=30),\n",
       " Row((age * 2)=38, abs(age)=19)]"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.selectExpr(\"age * 2\", \"abs(age)\").collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#### sort"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(age=30, name=u'Andy'),\n",
       " Row(age=19, name=u'Justin'),\n",
       " Row(age=None, name=u'Michael')]"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.sort(df.age.desc()).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(age=30, name=u'Andy'),\n",
       " Row(age=19, name=u'Justin'),\n",
       " Row(age=None, name=u'Michael')]"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.orderBy(df.age.desc()).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(age=None, name=u'Michael'),\n",
       " Row(age=19, name=u'Justin'),\n",
       " Row(age=30, name=u'Andy')]"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.sort(asc(\"age\")).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#### withColumn"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(age=None, name=u'Michael', age2=None),\n",
       " Row(age=30, name=u'Andy', age2=32),\n",
       " Row(age=19, name=u'Justin', age2=21)]"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.withColumn('age2', df.age + 2).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "##### withColumnRenamed"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(age2=None, name=u'Michael'),\n",
       " Row(age2=30, name=u'Andy'),\n",
       " Row(age2=19, name=u'Justin')]"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.withColumnRenamed('age', 'age2').collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### storagelevel\n",
    "#### subtract\n",
    "#### take\n",
    "#### union\n",
    "#### unpersist"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#### toDF"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(f1=None, f2=u'Michael'), Row(f1=30, f2=u'Andy'), Row(f1=19, f2=u'Justin')]"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.toDF('f1', 'f2').collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#### toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>age</th>\n",
       "      <th>name</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>NaN</td>\n",
       "      <td>Michael</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>30.0</td>\n",
       "      <td>Andy</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>19.0</td>\n",
       "      <td>Justin</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "    age     name\n",
       "0   NaN  Michael\n",
       "1  30.0     Andy\n",
       "2  19.0   Justin"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}