Streaming_CSV.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": [],
      "collapsed_sections": [],
      "authorship_tag": "ABX9TyOtyPbrmb0Vg5KFkKMvGQ8h",
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/momo54/large_scale_data_management/blob/main/Streaming_CSV.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Playing with structured input\n",
        "\n",
        "Structured Stream PRocessing on CSV data..."
      ],
      "metadata": {
        "id": "NQpoqd1WoHho"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "#install software\n",
        "# No clustering for this install !!\n",
        "!pip install pyspark\n",
        "!pip install -q findspark\n",
        "import findspark\n",
        "findspark.init()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "kYlmH__UmD8D",
        "outputId": "224414e3-52b7-446e-aaf2-33dff092c800"
      },
      "execution_count": 1,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Looking in indexes: https://pypi.org/simple, https://us-python.pkg.dev/colab-wheels/public/simple/\n",
            "Collecting pyspark\n",
            "  Downloading pyspark-3.3.0.tar.gz (281.3 MB)\n",
            "\u001b[K     |████████████████████████████████| 281.3 MB 47 kB/s \n",
            "\u001b[?25hCollecting py4j==0.10.9.5\n",
            "  Downloading py4j-0.10.9.5-py2.py3-none-any.whl (199 kB)\n",
            "\u001b[K     |████████████████████████████████| 199 kB 47.4 MB/s \n",
            "\u001b[?25hBuilding wheels for collected packages: pyspark\n",
            "  Building wheel for pyspark (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "  Created wheel for pyspark: filename=pyspark-3.3.0-py2.py3-none-any.whl size=281764026 sha256=429638ae9fdee2d63ce7dd641e5cee277d3e3818ad9495354d8e3e0dab53dbcb\n",
            "  Stored in directory: /root/.cache/pip/wheels/7a/8e/1b/f73a52650d2e5f337708d9f6a1750d451a7349a867f928b885\n",
            "Successfully built pyspark\n",
            "Installing collected packages: py4j, pyspark\n",
            "Successfully installed py4j-0.10.9.5 pyspark-3.3.0\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 2,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "ud5r6WuPl7av",
        "outputId": "5484bce9-cf73-46ff-bf1a-4b3ff1f62dd7"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "--2022-10-20 04:16:40--  https://raw.githubusercontent.com/lawlesst/vivo-sample-data/master/data/csv/people.csv\n",
            "Resolving raw.githubusercontent.com (raw.githubusercontent.com)... 185.199.108.133, 185.199.109.133, 185.199.110.133, ...\n",
            "Connecting to raw.githubusercontent.com (raw.githubusercontent.com)|185.199.108.133|:443... connected.\n",
            "HTTP request sent, awaiting response... 200 OK\n",
            "Length: 4095 (4.0K) [text/plain]\n",
            "Saving to: ‘people.csv’\n",
            "\n",
            "people.csv          100%[===================>]   4.00K  --.-KB/s    in 0s      \n",
            "\n",
            "2022-10-20 04:16:40 (52.3 MB/s) - ‘people.csv’ saved [4095/4095]\n",
            "\n"
          ]
        }
      ],
      "source": [
        "# get data\n",
        "#from https://che-kulhan.medium.com/how-to-use-pyspark-streaming-with-google-colaboratory-d08ded30cabf\n",
        "!wget https://raw.githubusercontent.com/lawlesst/vivo-sample-data/master/data/csv/people.csv"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "!split -l 10 people.csv people-\n",
        "!mkdir people"
      ],
      "metadata": {
        "id": "xiBudi_qmQ74"
      },
      "execution_count": 4,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# start the spark session\n",
        "\n",
        "from pyspark.sql import SparkSession\n",
        "from pyspark.sql.functions import explode\n",
        "from pyspark.sql.functions import split\n",
        "\n",
        "spark = SparkSession \\\n",
        "    .builder \\\n",
        "    .appName(\"Structured input\") \\\n",
        "    .getOrCreate()"
      ],
      "metadata": {
        "id": "Pom7dG_BmN72"
      },
      "execution_count": 29,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "from pyspark.sql.types import *\n",
        "schema = StructType([\n",
        "    StructField(\"person_ID\",IntegerType(),True), \n",
        "    StructField(\"last\", StringType(), True),\n",
        "    StructField(\"middle\", StringType(), True),\n",
        "    StructField(\"email\", StringType(), True),\n",
        "    StructField(\"phone\", StringType(), True),\n",
        "    StructField(\"fax\", StringType(), True),\n",
        "    StructField(\"title\", StringType(), True)])"
      ],
      "metadata": {
        "id": "rACspCHymZ0C"
      },
      "execution_count": 30,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "people_df = spark.readStream.format(\"csv\")\\\n",
        "   .schema(schema) \\\n",
        "   .load(\"people\")\n",
        "\n",
        "print(people_df.isStreaming)"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "aihWbC0-mac_",
        "outputId": "3008a263-081e-4e22-ed96-aa3a67b91df5"
      },
      "execution_count": 31,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "True\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "Ok, previous task should run now and never end until you stop it...\n",
        "Open the file view in the notebook and move people files to the people directory.\n",
        "\n",
        "This triggers the streaming process. Results appears in the result directory.\n",
        "\n",
        "When you understood. Interrupt the stream processing and clean running queries..."
      ],
      "metadata": {
        "id": "-VIPV5ROnbBK"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "\n",
        "#results_df = people_df.select(\"last\").where(\"length(middle)>0\")\n",
        "results_df = people_df.select(\"last\")\n",
        "query2 = results_df.writeStream \\\n",
        "  .format(\"json\") \\\n",
        "  .queryName(\"selectTable\") \\\n",
        "  .option(\"checkpointLocation\", \"checkpoint\") \\\n",
        "  .option(\"path\", \"results\") \\\n",
        "  .outputMode(\"append\") \\\n",
        "  .start() \n",
        "\n",
        "query2.awaitTermination()\n"
      ],
      "metadata": {
        "id": "Zscb4sd5me05"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [],
      "metadata": {
        "id": "LPGg5ZO-oA1X"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "for q in spark.streams.active:\n",
        "  print(\"stopping\",q.name)\n",
        "  q.stop()\n",
        "!rm -rf results/\n"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "AjbjNd2ymc73",
        "outputId": "7833840d-d26b-435a-8d97-d1dd1f1e5beb"
      },
      "execution_count": 33,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "stopping selectTable\n"
          ]
        }
      ]
    }
  ]
}