apache-spark/working-with-json-data-in-apache-spark.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Import Reuired Libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import *"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create SparkSession and SparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Sample Data"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "{\n",
    "\t\"id\": \"0001\",\n",
    "\t\"type\": \"donut\",\n",
    "\t\"name\": \"Cake\",\n",
    "\t\"ppu\": 0.55,\n",
    "\t\"batters\":\n",
    "\t\t{\n",
    "\t\t\t\"batter\":\n",
    "\t\t\t\t[\n",
    "\t\t\t\t\t{ \"id\": \"1001\", \"type\": \"Regular\" },\n",
    "\t\t\t\t\t{ \"id\": \"1002\", \"type\": \"Chocolate\" },\n",
    "\t\t\t\t\t{ \"id\": \"1003\", \"type\": \"Blueberry\" },\n",
    "\t\t\t\t\t{ \"id\": \"1004\", \"type\": \"Devil's Food\" }\n",
    "\t\t\t\t]\n",
    "\t\t},\n",
    "\t\"topping\":\n",
    "\t\t[\n",
    "\t\t\t{ \"id\": \"5001\", \"type\": \"None\" },\n",
    "\t\t\t{ \"id\": \"5002\", \"type\": \"Glazed\" },\n",
    "\t\t\t{ \"id\": \"5005\", \"type\": \"Sugar\" },\n",
    "\t\t\t{ \"id\": \"5007\", \"type\": \"Powdered Sugar\" },\n",
    "\t\t\t{ \"id\": \"5006\", \"type\": \"Chocolate with Sprinkles\" },\n",
    "\t\t\t{ \"id\": \"5003\", \"type\": \"Chocolate\" },\n",
    "\t\t\t{ \"id\": \"5004\", \"type\": \"Maple\" }\n",
    "\t\t]\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Read Sample JSON File"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [],
   "source": [
    "rawDF = spark.read.json(\"../data/sample.json\", multiLine = \"true\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Explore DataFrame Schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- batters: struct (nullable = true)\n",
      " |    |-- batter: array (nullable = true)\n",
      " |    |    |-- element: struct (containsNull = true)\n",
      " |    |    |    |-- id: string (nullable = true)\n",
      " |    |    |    |-- type: string (nullable = true)\n",
      " |-- id: string (nullable = true)\n",
      " |-- name: string (nullable = true)\n",
      " |-- ppu: double (nullable = true)\n",
      " |-- topping: array (nullable = true)\n",
      " |    |-- element: struct (containsNull = true)\n",
      " |    |    |-- id: string (nullable = true)\n",
      " |    |    |-- type: string (nullable = true)\n",
      " |-- type: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "rawDF.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Convert \"batters\" Nested Structure to Simple DataFrame"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Rename ID to Key"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {},
   "outputs": [],
   "source": [
    "sampleDF = rawDF.withColumnRenamed(\"id\", \"key\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Select \"batters\" columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- key: string (nullable = true)\n",
      " |-- batter: array (nullable = true)\n",
      " |    |-- element: struct (containsNull = true)\n",
      " |    |    |-- id: string (nullable = true)\n",
      " |    |    |-- type: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "batDF = sampleDF.select(\"key\", \"batters.batter\")\n",
    "batDF.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----------------------------------------------------------------------------+\n",
      "|key |batter                                                                       |\n",
      "+----+-----------------------------------------------------------------------------+\n",
      "|0001|[[1001, Regular], [1002, Chocolate], [1003, Blueberry], [1004, Devil's Food]]|\n",
      "+----+-----------------------------------------------------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "batDF.show(1, False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Creating a row for each element : explode"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+--------------------+\n",
      "| key|          new_batter|\n",
      "+----+--------------------+\n",
      "|0001|     [1001, Regular]|\n",
      "|0001|   [1002, Chocolate]|\n",
      "|0001|   [1003, Blueberry]|\n",
      "|0001|[1004, Devil's Food]|\n",
      "+----+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "bat2DF = batDF.select(\"key\", explode(\"batter\").alias(\"new_batter\"))\n",
    "bat2DF.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- key: string (nullable = true)\n",
      " |-- new_batter: struct (nullable = true)\n",
      " |    |-- id: string (nullable = true)\n",
      " |    |-- type: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "bat2DF.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+----+------------+\n",
      "| key|  id|        type|\n",
      "+----+----+------------+\n",
      "|0001|1001|     Regular|\n",
      "|0001|1002|   Chocolate|\n",
      "|0001|1003|   Blueberry|\n",
      "|0001|1004|Devil's Food|\n",
      "+----+----+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "bat2DF.select(\"key\", \"new_batter.*\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Creating a row for each struct element : explode"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+------+------------+\n",
      "| key|bat_id|    bat_type|\n",
      "+----+------+------------+\n",
      "|0001|  1001|     Regular|\n",
      "|0001|  1002|   Chocolate|\n",
      "|0001|  1003|   Blueberry|\n",
      "|0001|  1004|Devil's Food|\n",
      "+----+------+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "finalBatDF = (sampleDF\n",
    "        .select(\"key\", explode(\"batters.batter\").alias(\"new_batter\"))\n",
    "        .select(\"key\", \"new_batter.*\")\n",
    "        .withColumnRenamed(\"id\", \"bat_id\")\n",
    "        .withColumnRenamed(\"type\", \"bat_type\"))\n",
    "finalBatDF.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Select element from Array : topping"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+--------+\n",
      "|top_id|top_type|\n",
      "+------+--------+\n",
      "|  5001|    None|\n",
      "+------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sampleDF.select(col(\"topping\").getItem(0).id.alias(\"top_id\"), col(\"topping\").getItem(0).type.alias(\"top_type\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Convert \"toppings\" nested structure to simple DataFrame"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Creating a row for each Array element : explode"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+------+------------------------+\n",
      "|key |top_id|top_type                |\n",
      "+----+------+------------------------+\n",
      "|0001|5001  |None                    |\n",
      "|0001|5002  |Glazed                  |\n",
      "|0001|5005  |Sugar                   |\n",
      "|0001|5007  |Powdered Sugar          |\n",
      "|0001|5006  |Chocolate with Sprinkles|\n",
      "|0001|5003  |Chocolate               |\n",
      "|0001|5004  |Maple                   |\n",
      "+----+------+------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "topDF = (sampleDF\n",
    "        .select(\"key\", explode(\"topping\").alias(\"new_topping\"))\n",
    "        .select(\"key\", \"new_topping.*\")\n",
    "        .withColumnRenamed(\"id\", \"top_id\")\n",
    "        .withColumnRenamed(\"type\", \"top_type\")\n",
    "        )\n",
    "topDF.show(10, False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Thank You"
   ]
  }
 ],
 "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.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}