SparkSQL-Basic.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# From session create dataframe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+\n",
      "| age|   name|\n",
      "+----+-------+\n",
      "|null|Michael|\n",
      "|  30|   Andy|\n",
      "|  19| Justin|\n",
      "+----+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark = SparkSession.builder.getOrCreate()\n",
    "\n",
    "df = spark.read.json(\"file:///home/lygbug666/software/spark/examples/src/main/resources/people.json\")\n",
    "\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Dataframe operations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- age: long (nullable = true)\n",
      " |-- name: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+---------+\n",
      "|   name|(age + 1)|\n",
      "+-------+---------+\n",
      "|Michael|     null|\n",
      "|   Andy|       31|\n",
      "| Justin|       20|\n",
      "+-------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(df.name, df.age+1).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+----+\n",
      "|age|name|\n",
      "+---+----+\n",
      "| 30|Andy|\n",
      "+---+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.filter(df.age>20).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+\n",
      "| age|count|\n",
      "+----+-----+\n",
      "|  19|    1|\n",
      "|null|    1|\n",
      "|  30|    1|\n",
      "+----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy(df.age).count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----+\n",
      "|   name|count|\n",
      "+-------+-----+\n",
      "|Michael|    1|\n",
      "|   Andy|    1|\n",
      "| Justin|    1|\n",
      "+-------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy(\"name\").count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+\n",
      "| age|   name|\n",
      "+----+-------+\n",
      "|  30|   Andy|\n",
      "|  19| Justin|\n",
      "|null|Michael|\n",
      "+----+-------+\n",
      "\n",
      "+----+-------+\n",
      "| age|   name|\n",
      "+----+-------+\n",
      "|null|Michael|\n",
      "|  19| Justin|\n",
      "|  30|   Andy|\n",
      "+----+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.sort(df.age.desc()).show()\n",
    "df.sort(df.age).show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+\n",
      "| age|   name|\n",
      "+----+-------+\n",
      "|  30|   Andy|\n",
      "|  19| Justin|\n",
      "|null|Michael|\n",
      "+----+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.sort(df.age.desc(), df.name.asc()).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+----+\n",
      "|username| age|\n",
      "+--------+----+\n",
      "| Michael|null|\n",
      "|    Andy|  30|\n",
      "|  Justin|  19|\n",
      "+--------+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(df.name.alias(\"username\"),df.age).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Running SQL Queries Programmatically"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+\n",
      "| age|   name|\n",
      "+----+-------+\n",
      "|null|Michael|\n",
      "|  30|   Andy|\n",
      "|  19| Justin|\n",
      "+----+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.createOrReplaceTempView(\"people\")\n",
    "\n",
    "sqlDF = spark.sql(\"SELECT * FROM people\")\n",
    "\n",
    "sqlDF.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Global temporart view"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+\n",
      "| age|   name|\n",
      "+----+-------+\n",
      "|null|Michael|\n",
      "|  30|   Andy|\n",
      "|  19| Justin|\n",
      "+----+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.createGlobalTempView(\"people\")\n",
    "spark.sql(\"SELECT * FROM global_temp.people\").show()\n",
    "spark.newSession().sql(\"SELECT * FROM global_temp.people\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# RDD to DataFrames"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Reflection"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Michael, 29', 'Andy, 30', 'Justin, 19']"
      ]
     },
     "execution_count": 61,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc.textFile(\"file:///home/lygbug666/software/spark/examples/src/main/resources/people.txt\").collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql.types import Row"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def f(x):\n",
    "...     rel = {}\n",
    "...     rel['name'] = x[0]\n",
    "...     rel['age'] = x[1]\n",
    "...     return rel"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "peopleDF = sc.textFile(\"file:///home/lygbug666/software/spark/examples/src/main/resources/people.txt\").map(lambda line : line.split(',')).map(lambda x: Row(**f(x))).toDF()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {},
   "outputs": [],
   "source": [
    "peopleDF.createGlobalTempView(\"people2\")  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# 必须注册为临时表才能供下面的查询使用"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "metadata": {},
   "outputs": [],
   "source": [
    "personsDF = spark.sql(\"select * from global_temp.people2\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-------+\n",
      "|age|   name|\n",
      "+---+-------+\n",
      "| 29|Michael|\n",
      "| 30|   Andy|\n",
      "| 19| Justin|\n",
      "+---+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "personsDF.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Name: 29,Age:Michael', 'Name: 30,Age:Andy', 'Name: 19,Age:Justin']"
      ]
     },
     "execution_count": 71,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "personsDF.rdd.map(lambda t : \"Name:\"+t[0]+\",\"+\"Age:\"+t[1]).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# must be str, not NoneType"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 74,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "lines = sc.textFile(\"file:///home/lygbug666/software/spark/examples/src/main/resources/people.txt\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "parts = lines.map(lambda x : x.split(','))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 76,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "peopleRow = parts.map(lambda x : Row(name=x[0],age=int(x[1])))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "peopleschema = spark.createDataFrame(peopleRow)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "metadata": {},
   "outputs": [],
   "source": [
    "peopleschema.createOrReplaceTempView(\"peopletable\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "personDF = spark.sql(\"select * from peopletable\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 81,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-------+\n",
      "|age|   name|\n",
      "+---+-------+\n",
      "| 29|Michael|\n",
      "| 30|   Andy|\n",
      "| 19| Justin|\n",
      "+---+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "personDF.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 85,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "teenagers = spark.sql(\"SELECT name FROM peopletable WHERE age >= 13\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 86,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "teenNames = teenagers.rdd.map(lambda p: \"Name: \" + p.name).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 87,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Name: Michael\n",
      "Name: Andy\n",
      "Name: Justin\n"
     ]
    }
   ],
   "source": [
    "for name in teenNames:\n",
    "    print(name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## create schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 101,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 102,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "lines = sc.textFile(\"file:///home/lygbug666/software/spark/examples/src/main/resources/people.txt\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 103,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "parts = lines.map(lambda l: l.split(\",\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 104,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "people = parts.map(lambda p: (p[0], p[1].strip()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 105,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "schemaString = \"name age\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 106,
   "metadata": {},
   "outputs": [],
   "source": [
    "fields = [StructField(field_name, StringType(), True) for field_name in schemaString.split()]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 107,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql.types import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 108,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "peopleRDD = sc.textFile(\"file:///home/lygbug666/software/spark/examples/src/main/resources/people.txt\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 109,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "schemaString = \"name age\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 110,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "fields = list(map( lambda fieldName : StructField(fieldName, StringType(), nullable = True), schemaString.split(\" \")))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 111,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "schema = StructType(fields)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 113,
   "metadata": {},
   "outputs": [],
   "source": [
    "rowRDD = peopleRDD.map(lambda line : line.split(',')).map(lambda attributes : Row(attributes[0], attributes[1]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 114,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "peopleDF = spark.createDataFrame(rowRDD, schema)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 115,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "peopleDF.createOrReplaceTempView(\"people3\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 116,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "results = spark.sql(\"SELECT * FROM people3\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 118,
   "metadata": {},
   "outputs": [],
   "source": [
    "result = results.rdd.map( lambda attributes : \"name: \" + attributes[0]+\",\"+\"age:\"+attributes[1]).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 119,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "name: Michael,age: 29\n",
      "name: Andy,age: 30\n",
      "name: Justin,age: 19\n"
     ]
    }
   ],
   "source": [
    "for x in result:\n",
    "    print(x)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 125,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "name: Michael,age: 29\n",
      "name: Andy,age: 30\n",
      "name: Justin,age: 19\n"
     ]
    }
   ],
   "source": [
    "for x in results.rdd.map( lambda attributes : \"name: \" + attributes[0]+\",\"+\"age:\"+attributes[1]).collect():\n",
    "    print(x)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# DataSouce"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 127,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "peopleRDD = sc.textFile(\"file:///home/lygbug666/software/spark/examples/src/main/resources/people.txt\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 152,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.read.load(\"file:///home/lygbug666/software/spark/examples/src/main/resources/users.parquet\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 153,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "df.createOrReplaceTempView(\"parquetFile\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 154,
   "metadata": {},
   "outputs": [],
   "source": [
    "namesdf = spark.sql(\"select * from parquetFile\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 155,
   "metadata": {},
   "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": [
    "namesdf.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 146,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Alyssa', 'Ben']"
      ]
     },
     "execution_count": 146,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "    namesdf.rdd.map(lambda person : person.name).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 150,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "df = spark.sql(\"SELECT * FROM parquet.`file:///home/lygbug666/software/spark/examples/src/main/resources/users.parquet`\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 151,
   "metadata": {
    "scrolled": 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": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 149,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.read.text(\"file:///home/lygbug666/software/spark/examples/src/main/resources/people.txt\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}