jupyter/notebook/pyspark-dataframe-overview.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# pySpark Commands Reference\n",
    "https://spark.apache.org/docs/2.4.4/api/python/index.html\n",
    "\n",
    "\n",
    "## Connect to Spark Cluster\n",
    "https://github.com/minrk/findspark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder.appName(\"pyspark-df-overview\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.version"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create Spark DataFrame\n",
    "Dataset from: https://www.kaggle.com/uciml/adult-census-income/home"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.read.csv(\"/dataset/pyspark-df-overview/census_income.csv.gz\", header=True)\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Define a schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark.sql.types as t\n",
    "\n",
    "census_schema = t.StructType([\n",
    "      t.StructField('age', t.IntegerType(), True)\n",
    "    , t.StructField('workclass', t.StringType(), True)\n",
    "    , t.StructField('fnlwgt', t.IntegerType(), True)\n",
    "    , t.StructField('education', t.StringType(), True)\n",
    "    , t.StructField('education-num', t.IntegerType(), True)\n",
    "    , t.StructField('marital-status', t.StringType(), True)\n",
    "    , t.StructField('occupation', t.StringType(), True)\n",
    "    , t.StructField('relationship', t.StringType(), True)\n",
    "    , t.StructField('race', t.StringType(), True)\n",
    "    , t.StructField('sex', t.StringType(), True)\n",
    "    , t.StructField('capital-gain', t.DoubleType(), True)\n",
    "    , t.StructField('capital-loss', t.DoubleType(), True)\n",
    "    , t.StructField('hours-per-week', t.IntegerType(), True)\n",
    "    , t.StructField('native-country', t.StringType(), True)\n",
    "    , t.StructField('label', t.StringType(), True)\n",
    "])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Support for compressed (gziped) payload\n",
    "df = spark.read.csv(\"/dataset/pyspark-df-overview/census_income.csv.gz\", header=True, schema=census_schema)\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Drop unused column"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.drop('fnlwgt')\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Few operations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import count, avg, desc\n",
    "\n",
    "df.groupBy(['education']). \\\n",
    "agg(\n",
    "    count('*').alias('qty'), \n",
    "    avg('age').alias('avg_age')\n",
    ").orderBy(desc('qty')). \\\n",
    "show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Using SQL\n",
    "Same operation with SQL syntax"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.createOrReplaceTempView(\"census\")\n",
    "s = spark.sql(\"\"\"\n",
    "SELECT \n",
    "    education, \n",
    "    COUNT(*) AS qty, \n",
    "    AVG(age) AS avg_age\n",
    "FROM census\n",
    "GROUP BY education\n",
    "\"\"\")\n",
    "s.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# a transformation can be exposed as function\n",
    "def my_query(field):\n",
    "    return df.groupBy([field]). \\\n",
    "    agg(\n",
    "        count('*').alias('qty'), \n",
    "        avg('age').alias('avg_age')\n",
    "    ).orderBy(desc('qty'))\n",
    "    \n",
    "\n",
    "    \n",
    "print(my_query('workclass').show())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.select('age', 'education-num', 'capital-gain', 'capital-loss', 'hours-per-week').describe().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.select('workclass', 'education', 'marital-status').describe().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.freqItems(['marital-status']).show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.crosstab('age', 'label').sort(\"age_label\").show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.groupby('native-country').agg({'native-country': 'count'}).sort('count(native-country)').show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Check if there is missing data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import isnan, when, count, col\n",
    "\n",
    "# All columns\n",
    "# cols = df.columns\n",
    "# Selected columns\n",
    "cols = ['workclass', 'education-num', 'occupation', 'hours-per-week', 'native-country']\n",
    "\n",
    "# https://stackoverflow.com/a/44631639/570393\n",
    "df.select([count(when(isnan(c) | col(c).isNull(), c)).alias(c) for c in cols]).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Remove rows with missing data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Total rows\n",
    "print('total rows: %s' % df.count())\n",
    "\n",
    "# After droping NA records\n",
    "print('only complete rows: %s' % df.dropna().count())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Fill rows that contains missing data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def show_df(df, field='occupation'):\n",
    "    df.groupBy(field).count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "show_df(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Fill with a fixed value\n",
    "new_df = df.fillna({'occupation': 'Other-service'})\n",
    "\n",
    "# Count \n",
    "show_df(new_df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Better way\n",
    "\n",
    "Calc the `mean()` value of a column and use it on missing values.\n",
    "Also use a static string for categorical data "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import mean\n",
    "df.groupBy().agg(mean('hours-per-week').alias('hours-per-week')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import mean\n",
    "import pandas as pd\n",
    "\n",
    "data_to_fill = \\\n",
    "    df.groupBy().agg(mean('hours-per-week').alias('hours-per-week')).toPandas().to_dict('records')[0]\n",
    "\n",
    "# Simple Python Dict Update\n",
    "data_to_fill.update({'occupation': 'Other-service'})\n",
    "\n",
    "data_to_fill"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.fillna(data_to_fill).select('hours-per-week', 'occupation').show(50)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Creating charts with pandas & matplotlib\n",
    "https://pandas.pydata.org/pandas-docs/stable/api.html#api-dataframe-plotting\n",
    "\n",
    "**Important:** possible only when data become small enough to driver program"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# This is distributed\n",
    "df_spark = df.groupBy('workclass').agg(count('*').alias('counts')).orderBy('counts')\n",
    "# df_spark.show()\n",
    "\n",
    "# This is running on driver\n",
    "df_wk = df_spark.toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Check Pandas DF content\n",
    "df_wk"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import matplotlib.pyplot as plt\n",
    "%matplotlib inline\n",
    "\n",
    "df_wk.plot.bar(x='workclass', y='counts', figsize=(20,6));"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Stop Drive Program\n",
    "Release resources from Spark Cluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Continue Learning\n",
    "\n",
    "* [Kaggle Learn](https://www.kaggle.com/learn/overview)\n",
    "* [PySpark Cookbook](https://www.safaribooksonline.com/library/view/pyspark-cookbook/9781788835367/)\n",
    "\n",
    "## Other references\n",
    "\n",
    "* [PySpark Tutorial for Beginners: Machine Learning Example](https://www.guru99.com/pyspark-tutorial.html)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "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.7.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}