{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "7387c507-b420-4441-822c-ed9ba805d95e",
   "metadata": {},
   "source": [
    "# Chapter 7: Load and Behold - Data loading, storage, file formats"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "5c7cd870-ac96-4f64-be0c-af900c800ccd",
   "metadata": {
    "tags": [
     "remove_cell"
    ]
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Requirement already satisfied: pyspark==4.0.0.dev2 in /Users/amanda.liu/anaconda3/envs/llm-spark/lib/python3.11/site-packages (4.0.0.dev2)\n",
      "Requirement already satisfied: py4j==0.10.9.7 in /Users/amanda.liu/anaconda3/envs/llm-spark/lib/python3.11/site-packages (from pyspark==4.0.0.dev2) (0.10.9.7)\n"
     ]
    }
   ],
   "source": [
    "!pip install pyspark==4.0.0.dev2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "43230845-2ff0-432b-8c7d-edf695b3de31",
   "metadata": {
    "tags": [
     "remove_cell"
    ]
   },
   "outputs": [
   ],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .appName(\"Data Loading and Storage Example\") \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "d4d0fad4-cd2f-4458-a3ed-5cd10ca21abb",
   "metadata": {},
   "source": [
    "This section covers how to read and write data in various formats using PySpark. You'll learn how to load data from common file types (e.g., CSV, JSON, Parquet, ORC) and store data efficiently."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "b99bd39b-c36d-470f-bf4a-922115e323f8",
   "metadata": {},
   "source": [
    "## Reading Data"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "1b7bf82a-7727-46fa-bc95-fd5590c64ccd",
   "metadata": {},
   "source": [
    "### 1.1 Reading CSV Files\n",
    "\n",
    "CSV is one of the most common formats for data exchange. Here's how to load a CSV file into a DataFrame:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "8b4a6a34-011d-481b-8575-6b1c1846e7c3",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-----------------+-----------------+\n",
      "|Employee ID|             Role|         Location|\n",
      "+-----------+-----------------+-----------------+\n",
      "|      19238|     Data Analyst|      Seattle, WA|\n",
      "|      19239|Software Engineer|      Seattle, WA|\n",
      "|      19240|    IT Specialist|      Seattle, WA|\n",
      "|      19241|     Data Analyst|     New York, NY|\n",
      "|      19242|        Recruiter|San Francisco, CA|\n",
      "|      19243|  Product Manager|     New York, NY|\n",
      "+-----------+-----------------+-----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "csv_df = spark.read.csv(\"../data/employees.csv\", header=True, inferSchema=True)\n",
    "csv_df.show()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "d8a98e55-b9d1-4cd7-a28c-b9738d55cae7",
   "metadata": {},
   "source": [
    "**Explanation:**\n",
    "- `header=True`: Treats the first line as column names.\n",
    "- `inferSchema=True`: Automatically infers data types of columns."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "302b8d74-33ea-4a54-95c3-760ce6b0f816",
   "metadata": {},
   "source": [
    "### 1.2 Reading JSON Files\n",
    "\n",
    "Loading JSON files is simple and allows you to handle both single-line and multi-line JSON structures:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "39e3fc27-a9bf-493a-8f89-6c087ed4e89b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-----------------+-----------------+\n",
      "|Employee ID|         Location|             Role|\n",
      "+-----------+-----------------+-----------------+\n",
      "|      19238|      Seattle, WA|     Data Analyst|\n",
      "|      19239|      Seattle, WA|Software Engineer|\n",
      "|      19240|      Seattle, WA|    IT Specialist|\n",
      "|      19241|     New York, NY|     Data Analyst|\n",
      "|      19242|San Francisco, CA|        Recruiter|\n",
      "|      19243|     New York, NY|  Product Manager|\n",
      "+-----------+-----------------+-----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "json_df = spark.read.option(\"multiline\", \"true\").json(\"../data/employees.json\")\n",
    "json_df.show()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "d4a46228-9f55-41a5-9e44-8a9efdac4985",
   "metadata": {},
   "source": [
    "**Explanation:**\n",
    "- `multiline=\"true\"`: Allows reading multi-line JSON structures."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "d0578c9a-90d5-462d-b35a-f05aceaf70d9",
   "metadata": {},
   "source": [
    "### 1.3 Reading Parquet Files\n",
    "\n",
    "Parquet is a columnar format that supports efficient data compression and encoding:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "d04aa812-cd22-439b-aea5-0a148ccc5d1b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-----------------+-----------------+\n",
      "|Employee ID|         Location|             Role|\n",
      "+-----------+-----------------+-----------------+\n",
      "|      19239|      Seattle, WA|Software Engineer|\n",
      "|      19243|     New York, NY|  Product Manager|\n",
      "|      19242|San Francisco, CA|        Recruiter|\n",
      "|      19241|     New York, NY|     Data Analyst|\n",
      "|      19240|      Seattle, WA|    IT Specialist|\n",
      "|      19238|      Seattle, WA|     Data Analyst|\n",
      "+-----------+-----------------+-----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "parquet_df = spark.read.parquet(\"../data/employees.parquet\")\n",
    "parquet_df.show()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "0c1a93b4-ed28-437f-a889-afb9fa128739",
   "metadata": {},
   "source": [
    "**Tip:** Parquet files are highly efficient for storing data due to columnar storage and compression."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "6442a9e1-f8f5-4b41-a5d3-f7983458aa30",
   "metadata": {},
   "source": [
    "### 1.4 Reading ORC Files\n",
    "\n",
    "ORC is another columnar file format, often used in Hadoop environments:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "3af892a9-1a1b-400b-84ec-e0f787689d2c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-----------------+-----------------+\n",
      "|Employee ID|         Location|             Role|\n",
      "+-----------+-----------------+-----------------+\n",
      "|      19242|San Francisco, CA|        Recruiter|\n",
      "|      19239|      Seattle, WA|Software Engineer|\n",
      "|      19240|      Seattle, WA|    IT Specialist|\n",
      "|      19243|     New York, NY|  Product Manager|\n",
      "|      19238|      Seattle, WA|     Data Analyst|\n",
      "|      19241|     New York, NY|     Data Analyst|\n",
      "+-----------+-----------------+-----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "orc_df = spark.read.orc(\"../data/employees.orc\")\n",
    "orc_df.show()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "c19666a4-a31f-4f76-b328-15b8d476f7e3",
   "metadata": {},
   "source": [
    "## Writing Data"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "20f3344d-0617-458e-86b5-ced402b20d89",
   "metadata": {},
   "source": [
    "### 2.1 Writing Data as CSV"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "b9961491-2b46-4252-9aae-014aadd98d4c",
   "metadata": {},
   "outputs": [],
   "source": [
    "csv_df.write.csv(\"../data/employees_out.csv\", mode=\"overwrite\", header=True)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "6227bb33-01b9-4f4b-a17a-4c9678371192",
   "metadata": {},
   "source": [
    "**Explanation:**\n",
    "- `mode=\"overwrite\"`: If the directory exists, it will be replaced.\n",
    "- `header=True`: Writes the column names as the first line."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "f3151eab-0db9-4087-8df7-c36995c14243",
   "metadata": {},
   "source": [
    "### 2.2 Writing Data as Parquet\n",
    "\n",
    "Parquet format is recommended for large datasets:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "ec2b8563-f681-4cbc-93e8-1da134a51008",
   "metadata": {},
   "outputs": [],
   "source": [
    "parquet_df.write.parquet(\"../data/employees_out.parquet\", mode=\"overwrite\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "988990e8-6e62-4410-945a-d284699871d6",
   "metadata": {},
   "source": [
    "### 2.3 Writing Data as ORC"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "e4827d27-1b41-4661-9dd6-efff6bea2c16",
   "metadata": {},
   "outputs": [],
   "source": [
    "json_df.write.orc(\"../data/employees_out.orc\", mode=\"overwrite\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "ec497f26-95b4-4a4d-be3a-5b51576c2e98",
   "metadata": {},
   "source": [
    "**Tip:** Parquet and ORC formats are best for efficient storage and quick reads."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "4e23c3bf-a365-41c7-84cd-3d70589eb051",
   "metadata": {},
   "source": [
    "## Additional Options and Configurations\n",
    "\n",
    "You can customize how data is read and written by using additional options. Here are a few examples:"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "71080e8a-2be3-4726-bb15-0a24d08df46e",
   "metadata": {},
   "source": [
    "### Custom Delimiter in CSV:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "4a2ea041-4564-422a-81f4-94a745f99848",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------------------+\n",
      "|_c0                                  |\n",
      "+-------------------------------------+\n",
      "|Employee ID,Role,Location            |\n",
      "|19238,Data Analyst,\"Seattle, WA\"     |\n",
      "|19239,Software Engineer,\"Seattle, WA\"|\n",
      "|19240,IT Specialist,\"Seattle, WA\"    |\n",
      "|19241,Data Analyst,\"New York, NY\"    |\n",
      "|19242,Recruiter,\"San Francisco, CA\"  |\n",
      "|19243,Product Manager,\"New York, NY\" |\n",
      "+-------------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.read.option(\"delimiter\", \";\").csv(\"../data/employees.csv\").show(truncate=False)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "a93d8450-f675-4288-9cdf-7aa00f806136",
   "metadata": {},
   "source": [
    "### Handling Null Values:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "50069b43-c926-4453-8e37-6bc605d9bff2",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-----------------+-----------------+\n",
      "|_c0        |_c1              |_c2              |\n",
      "+-----------+-----------------+-----------------+\n",
      "|Employee ID|Role             |Location         |\n",
      "|19238      |Data Analyst     |Seattle, WA      |\n",
      "|19239      |Software Engineer|Seattle, WA      |\n",
      "|19240      |IT Specialist    |Seattle, WA      |\n",
      "|19241      |Data Analyst     |New York, NY     |\n",
      "|19242      |Recruiter        |San Francisco, CA|\n",
      "|19243      |Product Manager  |New York, NY     |\n",
      "+-----------+-----------------+-----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.read.option(\"nullValue\", \"NULL\").csv(\"../data/employees.csv\").show(truncate=False)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "0379e733-08b6-4851-9750-117ce4bfca09",
   "metadata": {},
   "source": [
    "### Compression Options:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "d0a2d239-6e87-4ea0-93b5-7272e77cedf8",
   "metadata": {},
   "outputs": [],
   "source": [
    "parquet_df.write.option(\"compression\", \"gzip\").parquet(\"../data/employees_out.parquet\", mode=\"overwrite\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "202f4ee8-0c24-40c3-9e54-ee7eef52799e",
   "metadata": {},
   "source": [
    "See the [PySpark API reference](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/io.html) for Input/Output to check all supported functions and options."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "",
   "language": "python",
   "name": ""
  },
  "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.11.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}