spark_pipelining_withPCA.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": [],
      "collapsed_sections": [
        "Y3vw_quGSgrW",
        "yC40UHIKPJEL",
        "w8cVWRNoPOS6",
        "mfFBQ0FLcolo",
        "IJNKSScPcsDS",
        "Y159J8TDc3wS",
        "7M2kWg3dc6FT",
        "MYRPnQhTRdOn",
        "iHd3YFPHO1SV",
        "L6XYg1IGs17n",
        "QxLQzJD4p81p"
      ],
      "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/harnalashok/hadoop/blob/main/spark_pipelining_withPCA.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "4q64Z8IYb8fr"
      },
      "source": [
        "# Last amended: 04th April, 2023\n",
        "# Myfolder: github/hadoop\n",
        "# Objectives:\n",
        "#             i)  Install pyspark on colab\n",
        "#             ii) Perform ML operations using pipeline\n",
        "#                 (version installed is spark-3.3.2)\n",
        "#\n",
        "# Java 8 install: https://stackoverflow.com/a/58191107\n",
        "# Hadoop install: https://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-common/SingleCluster.html\n",
        "# Spark install:  https://stackoverflow.com/a/64183749\n",
        "#                 https://www.analyticsvidhya.com/blog/2020/11/a-must-read-guide-on-how-to-work-with-pyspark-on-google-colab-for-data-scientists/"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "GuuFSunM_rL2"
      },
      "source": [
        "# Spark Reference API\n",
        "a. [Spark ETL operations](https://spark.apache.org/docs/latest/ml-features.html#extracting-transforming-and-selecting-features) <br>\n",
        "b. Dataframe [APIs list](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/dataframe.html) at a glance<br>\n",
        "c. Spark [Classification and Regression](https://spark.apache.org/docs/latest/ml-classification-regression.html) <br>\n",
        "c. ALso look at useful [this source code](https://spark.apache.org/docs/latest/api/python/_modules/pyspark/sql/functions.html) of functions that has examples<br>\n",
        "\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Y3vw_quGSgrW"
      },
      "source": [
        "# A. Full spark install\n",
        "Installs `pyspark (spark-3.3.2-bin-hadoop3)` "
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "yC40UHIKPJEL"
      },
      "source": [
        "### 1.0 Libraries"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "HBVUrlkidyaE"
      },
      "source": [
        "# 1.0 How to set environment variable\n",
        "import os  \n",
        "import time  "
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "w8cVWRNoPOS6"
      },
      "source": [
        "## 2.0 Define some functions"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "mfFBQ0FLcolo"
      },
      "source": [
        "#### ssh_install()"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "mgS9HNCyR7C0"
      },
      "source": [
        "# 2.0 Function to install ssh client and sshd (Server)\n",
        "def ssh_install():\n",
        "  print(\"\\n--1. Download and install ssh server----\\n\")\n",
        "  ! sudo apt-get remove openssh-client openssh-server\n",
        "  ! sudo apt install openssh-client openssh-server\n",
        "  \n",
        "  print(\"\\n--2. Restart ssh server----\\n\")\n",
        "  ! service ssh restart"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "IJNKSScPcsDS"
      },
      "source": [
        "#### Java install"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "BsFu84PSR9jR"
      },
      "source": [
        "# 3.0 Function to download and install java 8\n",
        "def install_java():\n",
        "  ! rm -rf /usr/java\n",
        "\n",
        "  print(\"\\n--Download and install Java 8----\\n\")\n",
        "  !apt-get install -y openjdk-8-jdk-headless -qq > /dev/null        # install openjdk\n",
        "  os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-8-openjdk-amd64\"     # set environment variable\n",
        "\n",
        "  !update-alternatives --set java /usr/lib/jvm/java-8-openjdk-amd64/jre/bin/java\n",
        "  !update-alternatives --set javac /usr/lib/jvm/java-8-openjdk-amd64/bin/javac\n",
        "  \n",
        "  !mkdir -p /usr/java\n",
        "  ! ln -s \"/usr/lib/jvm/java-8-openjdk-amd64\"  \"/usr/java\"\n",
        "  ! mv \"/usr/java/java-8-openjdk-amd64\"  \"/usr/java/latest\"\n",
        "  \n",
        "  !java -version       #check java version\n",
        "  !javac -version"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Y159J8TDc3wS"
      },
      "source": [
        "#### setup ssh passphrase"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "EOFbfw7n0Pps"
      },
      "source": [
        "# 6.0 Function tp setup ssh passphrase\n",
        "def set_keys():\n",
        "  print(\"\\n---22. Generate SSH keys----\\n\")\n",
        "  ! cd ~ ; pwd \n",
        "  ! cd ~ ; ssh-keygen -t rsa -P '' -f ~/.ssh/id_rsa\n",
        "  ! cd ~ ; cat ~/.ssh/id_rsa.pub >> ~/.ssh/authorized_keys\n",
        "  ! cd ~ ; chmod 0600 ~/.ssh/authorized_keys\n"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "7M2kWg3dc6FT"
      },
      "source": [
        "#### Set environment"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "CRSn9XAV4rsR"
      },
      "source": [
        "# 7.0 Function to set up environmental variables\n",
        "def set_env():\n",
        "  print(\"\\n---23. Set Environment variables----\\n\")\n",
        "  # 'export' command does not work in colab\n",
        "  # https://stackoverflow.com/a/57240319\n",
        "  os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-8-openjdk-amd64\"     #set environment variable\n",
        "  os.environ[\"JRE_HOME\"] = \"/usr/lib/jvm/java-8-openjdk-amd64/jre\"   \n",
        "  "
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "WII-UNCzc9qJ"
      },
      "source": [
        "#### function to install prerequisites\n",
        "java and ssh<br>\n"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "oh1Mi0rHFpkU"
      },
      "source": [
        "# 8.0 Function to call all functions\n",
        "def install_components():\n",
        "  print(\"\\n--Install java----\\n\")\n",
        "  ssh_install()\n",
        "  install_java()  \n",
        "  #set_keys()\n",
        "  set_env()\n"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "iHJyHMhUdCRZ"
      },
      "source": [
        "## 3.0 Install components\n",
        "Start downloading, install and configure. Takes around 2 minutes<br>\n",
        "Your <u>input *'y'* is required </u>at one place while overwriting earlier ssh keys"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "77YQikvsJiTm",
        "outputId": "2ff6174c-caee-44bd-d710-0100d91f5533",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "source": [
        "# 9.0 Start installation\n",
        "start = time.time()\n",
        "install_components()\n",
        "end = time.time()\n",
        "print(\"\\n---Time taken----\\n\")\n",
        "print((end- start)/60)"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "\n",
            "--Install java----\n",
            "\n",
            "\n",
            "--1. Download and install ssh server----\n",
            "\n",
            "Reading package lists... Done\n",
            "Building dependency tree       \n",
            "Reading state information... Done\n",
            "Package 'openssh-server' is not installed, so not removed\n",
            "The following packages were automatically installed and are no longer required:\n",
            "  libboost-atomic-dev libboost-atomic1.71-dev libboost-atomic1.71.0\n",
            "  libboost-chrono-dev libboost-chrono1.71-dev libboost-chrono1.71.0\n",
            "  libboost-container-dev libboost-container1.71-dev libboost-container1.71.0\n",
            "  libboost-context-dev libboost-context1.71-dev libboost-context1.71.0\n",
            "  libboost-coroutine-dev libboost-coroutine1.71-dev libboost-coroutine1.71.0\n",
            "  libboost-date-time-dev libboost-date-time1.71-dev libboost-date-time1.71.0\n",
            "  libboost-exception-dev libboost-exception1.71-dev libboost-fiber-dev\n",
            "  libboost-fiber1.71-dev libboost-fiber1.71.0 libboost-filesystem-dev\n",
            "  libboost-filesystem1.71-dev libboost-filesystem1.71.0 libboost-graph-dev\n",
            "  libboost-graph-parallel-dev libboost-graph-parallel1.71-dev\n",
            "  libboost-graph-parallel1.71.0 libboost-graph1.71-dev libboost-graph1.71.0\n",
            "  libboost-iostreams-dev libboost-iostreams1.71-dev libboost-locale-dev\n",
            "  libboost-locale1.71-dev libboost-locale1.71.0 libboost-log-dev\n",
            "  libboost-log1.71-dev libboost-log1.71.0 libboost-math-dev\n",
            "  libboost-math1.71-dev libboost-math1.71.0 libboost-mpi1.71.0\n",
            "  libboost-numpy-dev libboost-numpy1.71-dev libboost-numpy1.71.0\n",
            "  libboost-program-options-dev libboost-program-options1.71-dev\n",
            "  libboost-program-options1.71.0 libboost-python-dev libboost-python1.71-dev\n",
            "  libboost-python1.71.0 libboost-random-dev libboost-random1.71-dev\n",
            "  libboost-random1.71.0 libboost-regex-dev libboost-regex1.71-dev\n",
            "  libboost-regex1.71.0 libboost-serialization-dev\n",
            "  libboost-serialization1.71-dev libboost-serialization1.71.0\n",
            "  libboost-stacktrace-dev libboost-stacktrace1.71-dev\n",
            "  libboost-stacktrace1.71.0 libboost-system-dev libboost-system1.71-dev\n",
            "  libboost-system1.71.0 libboost-test-dev libboost-test1.71-dev\n",
            "  libboost-test1.71.0 libboost-thread-dev libboost-thread1.71-dev\n",
            "  libboost-timer-dev libboost-timer1.71-dev libboost-timer1.71.0\n",
            "  libboost-tools-dev libboost-type-erasure-dev libboost-type-erasure1.71-dev\n",
            "  libboost-type-erasure1.71.0 libboost-wave-dev libboost-wave1.71-dev\n",
            "  libboost-wave1.71.0 libboost1.71-tools-dev libcbor0.6 libevent-dev\n",
            "  libevent-extra-2.1-7 libevent-openssl-2.1-7 libfido2-1 libhwloc-dev\n",
            "  libibverbs-dev libnl-3-dev libnl-route-3-dev libnuma-dev\n",
            "  libnvidia-common-525 openmpi-common python3-dev python3-distutils\n",
            "  python3-lib2to3 python3.8-dev\n",
            "Use 'sudo apt autoremove' to remove them.\n",
            "The following packages will be REMOVED:\n",
            "  libboost-all-dev libboost-mpi-dev libboost-mpi-python-dev\n",
            "  libboost-mpi-python1.71-dev libboost-mpi-python1.71.0 libboost-mpi1.71-dev\n",
            "  libopenmpi-dev mpi-default-bin mpi-default-dev openmpi-bin openssh-client\n",
            "0 upgraded, 0 newly installed, 11 to remove and 23 not upgraded.\n",
            "After this operation, 20.2 MB disk space will be freed.\n",
            "(Reading database ... 128293 files and directories currently installed.)\n",
            "Removing libboost-all-dev (1.71.0.0ubuntu2) ...\n",
            "Removing libboost-mpi-dev (1.71.0.0ubuntu2) ...\n",
            "Removing libboost-mpi-python-dev (1.71.0.0ubuntu2) ...\n",
            "Removing libboost-mpi-python1.71-dev (1.71.0-6ubuntu6) ...\n",
            "Removing libboost-mpi-python1.71.0 (1.71.0-6ubuntu6) ...\n",
            "Removing libboost-mpi1.71-dev (1.71.0-6ubuntu6) ...\n",
            "Removing mpi-default-dev (1.13) ...\n",
            "Removing libopenmpi-dev:amd64 (4.0.3-0ubuntu1) ...\n",
            "Removing mpi-default-bin (1.13) ...\n",
            "Removing openmpi-bin (4.0.3-0ubuntu1) ...\n",
            "Removing openssh-client (1:8.2p1-4ubuntu0.5) ...\n",
            "Processing triggers for man-db (2.9.1-1) ...\n",
            "Processing triggers for libc-bin (2.31-0ubuntu9.9) ...\n",
            "Reading package lists... Done\n",
            "Building dependency tree       \n",
            "Reading state information... Done\n",
            "The following packages were automatically installed and are no longer required:\n",
            "  libboost-atomic-dev libboost-atomic1.71-dev libboost-atomic1.71.0\n",
            "  libboost-chrono-dev libboost-chrono1.71-dev libboost-chrono1.71.0\n",
            "  libboost-container-dev libboost-container1.71-dev libboost-container1.71.0\n",
            "  libboost-context-dev libboost-context1.71-dev libboost-context1.71.0\n",
            "  libboost-coroutine-dev libboost-coroutine1.71-dev libboost-coroutine1.71.0\n",
            "  libboost-date-time-dev libboost-date-time1.71-dev libboost-date-time1.71.0\n",
            "  libboost-exception-dev libboost-exception1.71-dev libboost-fiber-dev\n",
            "  libboost-fiber1.71-dev libboost-fiber1.71.0 libboost-filesystem-dev\n",
            "  libboost-filesystem1.71-dev libboost-filesystem1.71.0 libboost-graph-dev\n",
            "  libboost-graph-parallel-dev libboost-graph-parallel1.71-dev\n",
            "  libboost-graph-parallel1.71.0 libboost-graph1.71-dev libboost-graph1.71.0\n",
            "  libboost-iostreams-dev libboost-iostreams1.71-dev libboost-locale-dev\n",
            "  libboost-locale1.71-dev libboost-locale1.71.0 libboost-log-dev\n",
            "  libboost-log1.71-dev libboost-log1.71.0 libboost-math-dev\n",
            "  libboost-math1.71-dev libboost-math1.71.0 libboost-mpi1.71.0\n",
            "  libboost-numpy-dev libboost-numpy1.71-dev libboost-numpy1.71.0\n",
            "  libboost-program-options-dev libboost-program-options1.71-dev\n",
            "  libboost-program-options1.71.0 libboost-python-dev libboost-python1.71-dev\n",
            "  libboost-python1.71.0 libboost-random-dev libboost-random1.71-dev\n",
            "  libboost-random1.71.0 libboost-regex-dev libboost-regex1.71-dev\n",
            "  libboost-regex1.71.0 libboost-serialization-dev\n",
            "  libboost-serialization1.71-dev libboost-serialization1.71.0\n",
            "  libboost-stacktrace-dev libboost-stacktrace1.71-dev\n",
            "  libboost-stacktrace1.71.0 libboost-system-dev libboost-system1.71-dev\n",
            "  libboost-system1.71.0 libboost-test-dev libboost-test1.71-dev\n",
            "  libboost-test1.71.0 libboost-thread-dev libboost-thread1.71-dev\n",
            "  libboost-timer-dev libboost-timer1.71-dev libboost-timer1.71.0\n",
            "  libboost-tools-dev libboost-type-erasure-dev libboost-type-erasure1.71-dev\n",
            "  libboost-type-erasure1.71.0 libboost-wave-dev libboost-wave1.71-dev\n",
            "  libboost-wave1.71.0 libboost1.71-tools-dev libevent-dev libevent-extra-2.1-7\n",
            "  libevent-openssl-2.1-7 libhwloc-dev libibverbs-dev libnl-3-dev\n",
            "  libnl-route-3-dev libnuma-dev libnvidia-common-525 openmpi-common\n",
            "  python3-dev python3-distutils python3-lib2to3 python3.8-dev\n",
            "Use 'sudo apt autoremove' to remove them.\n",
            "The following additional packages will be installed:\n",
            "  ncurses-term openssh-sftp-server python3-distro ssh-import-id\n",
            "Suggested packages:\n",
            "  keychain libpam-ssh monkeysphere ssh-askpass molly-guard ufw\n",
            "The following NEW packages will be installed:\n",
            "  ncurses-term openssh-client openssh-server openssh-sftp-server\n",
            "  python3-distro ssh-import-id\n",
            "0 upgraded, 6 newly installed, 0 to remove and 23 not upgraded.\n",
            "Need to get 1,373 kB of archives.\n",
            "After this operation, 10.3 MB of additional disk space will be used.\n",
            "Get:1 http://archive.ubuntu.com/ubuntu focal-updates/main amd64 openssh-client amd64 1:8.2p1-4ubuntu0.5 [671 kB]\n",
            "Get:2 http://archive.ubuntu.com/ubuntu focal/main amd64 ncurses-term all 6.2-0ubuntu2 [249 kB]\n",
            "Get:3 http://archive.ubuntu.com/ubuntu focal-updates/main amd64 openssh-sftp-server amd64 1:8.2p1-4ubuntu0.5 [51.5 kB]\n",
            "Get:4 http://archive.ubuntu.com/ubuntu focal-updates/main amd64 openssh-server amd64 1:8.2p1-4ubuntu0.5 [377 kB]\n",
            "Get:5 http://archive.ubuntu.com/ubuntu focal/main amd64 python3-distro all 1.4.0-1 [14.6 kB]\n",
            "Get:6 http://archive.ubuntu.com/ubuntu focal/main amd64 ssh-import-id all 5.10-0ubuntu1 [10.0 kB]\n",
            "Fetched 1,373 kB in 1s (1,722 kB/s)\n",
            "debconf: unable to initialize frontend: Dialog\n",
            "debconf: (No usable dialog-like program is installed, so the dialog based frontend cannot be used. at /usr/share/perl5/Debconf/FrontEnd/Dialog.pm line 76, <> line 6.)\n",
            "debconf: falling back to frontend: Readline\n",
            "debconf: unable to initialize frontend: Readline\n",
            "debconf: (This frontend requires a controlling tty.)\n",
            "debconf: falling back to frontend: Teletype\n",
            "dpkg-preconfigure: unable to re-open stdin: \n",
            "Selecting previously unselected package openssh-client.\n",
            "(Reading database ... 127453 files and directories currently installed.)\n",
            "Preparing to unpack .../0-openssh-client_1%3a8.2p1-4ubuntu0.5_amd64.deb ...\n",
            "Unpacking openssh-client (1:8.2p1-4ubuntu0.5) ...\n",
            "Selecting previously unselected package ncurses-term.\n",
            "Preparing to unpack .../1-ncurses-term_6.2-0ubuntu2_all.deb ...\n",
            "Unpacking ncurses-term (6.2-0ubuntu2) ...\n",
            "Selecting previously unselected package openssh-sftp-server.\n",
            "Preparing to unpack .../2-openssh-sftp-server_1%3a8.2p1-4ubuntu0.5_amd64.deb ...\n",
            "Unpacking openssh-sftp-server (1:8.2p1-4ubuntu0.5) ...\n",
            "Selecting previously unselected package openssh-server.\n",
            "Preparing to unpack .../3-openssh-server_1%3a8.2p1-4ubuntu0.5_amd64.deb ...\n",
            "Unpacking openssh-server (1:8.2p1-4ubuntu0.5) ...\n",
            "Selecting previously unselected package python3-distro.\n",
            "Preparing to unpack .../4-python3-distro_1.4.0-1_all.deb ...\n",
            "Unpacking python3-distro (1.4.0-1) ...\n",
            "Selecting previously unselected package ssh-import-id.\n",
            "Preparing to unpack .../5-ssh-import-id_5.10-0ubuntu1_all.deb ...\n",
            "Unpacking ssh-import-id (5.10-0ubuntu1) ...\n",
            "Setting up python3-distro (1.4.0-1) ...\n",
            "Setting up openssh-client (1:8.2p1-4ubuntu0.5) ...\n",
            "Setting up ssh-import-id (5.10-0ubuntu1) ...\n",
            "Attempting to convert /etc/ssh/ssh_import_id\n",
            "Setting up ncurses-term (6.2-0ubuntu2) ...\n",
            "Setting up openssh-sftp-server (1:8.2p1-4ubuntu0.5) ...\n",
            "Setting up openssh-server (1:8.2p1-4ubuntu0.5) ...\n",
            "debconf: unable to initialize frontend: Dialog\n",
            "debconf: (No usable dialog-like program is installed, so the dialog based frontend cannot be used. at /usr/share/perl5/Debconf/FrontEnd/Dialog.pm line 76.)\n",
            "debconf: falling back to frontend: Readline\n",
            "\n",
            "Creating config file /etc/ssh/sshd_config with new version\n",
            "Creating SSH2 RSA key; this may take some time ...\n",
            "3072 SHA256:ljEuqhjeg7mzYoZLhlfbKuUYjLyaT+tQ6jRPXcRPljI root@640341b500da (RSA)\n",
            "Creating SSH2 ECDSA key; this may take some time ...\n",
            "256 SHA256:IeQvqCcL70yTMoT+/+Dpt+pa/RB0RMyNIp8LCuBwa8E root@640341b500da (ECDSA)\n",
            "Creating SSH2 ED25519 key; this may take some time ...\n",
            "256 SHA256:0QKLw2os9EdNgbtgt4+LbuEyd2eCGvQVdSUSM/qB8Tc root@640341b500da (ED25519)\n",
            "Created symlink /etc/systemd/system/sshd.service → /lib/systemd/system/ssh.service.\n",
            "Created symlink /etc/systemd/system/multi-user.target.wants/ssh.service → /lib/systemd/system/ssh.service.\n",
            "invoke-rc.d: could not determine current runlevel\n",
            "invoke-rc.d: policy-rc.d denied execution of start.\n",
            "Processing triggers for man-db (2.9.1-1) ...\n",
            "Processing triggers for systemd (245.4-4ubuntu3.20) ...\n",
            "\n",
            "--2. Restart ssh server----\n",
            "\n",
            " * Restarting OpenBSD Secure Shell server sshd\n",
            "   ...done.\n",
            "\n",
            "--Download and install Java 8----\n",
            "\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/jre/bin/java to provide /usr/bin/java (java) in manual mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/javac to provide /usr/bin/javac (javac) in manual mode\n",
            "openjdk version \"1.8.0_362\"\n",
            "OpenJDK Runtime Environment (build 1.8.0_362-8u362-ga-0ubuntu1~20.04.1-b09)\n",
            "OpenJDK 64-Bit Server VM (build 25.362-b09, mixed mode)\n",
            "javac 1.8.0_362\n",
            "\n",
            "---23. Set Environment variables----\n",
            "\n",
            "\n",
            "---Time taken----\n",
            "\n",
            "0.4485221783320109\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "XBFVWWuafLBL"
      },
      "source": [
        "## 4.0 Install spark\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "CyuLCFRJsvW3"
      },
      "source": [
        "### Define functions"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "kk3BP0OfYpT1"
      },
      "source": [
        "`findspark`: PySpark isn't on `sys.path` by default, but that doesn't mean it can't be used as a regular library. You can address this by either symlinking pyspark into your site-packages, or adding `pyspark` to `sys.path` at runtime. `findspark` does the latter."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "opLGqtPRfM_5"
      },
      "source": [
        "# 1.0 Function to download and unzip spark\n",
        "def spark_koalas_install():\n",
        "  print(\"\\n--1.1 Install findspark----\\n\")\n",
        "  !pip install -q findspark\n",
        "\n",
        "  #print(\"\\n--1.2 Install databricks Koalas----\\n\")\n",
        "  #!pip install koalas\n",
        "  \n",
        "  # This download link NEEDS TO BE CHECKED AGAIN\n",
        "  print(\"\\n--1.3 Download Apache tar.gz----\\n\")\n",
        "            \n",
        "  ! wget -c https://dlcdn.apache.org/spark/spark-3.3.2/spark-3.3.2-bin-hadoop3.tgz \n",
        "\n",
        "  print(\"\\n--1.4 Transfer downloaded content and unzip tar.gz----\\n\")\n",
        "  !  mv /content/spark*   /opt/\n",
        "  ! tar -xzf /opt/spark-3.3.2-bin-hadoop3.tgz  --directory /opt/\n",
        "\n",
        "  print(\"\\n--1.5 Check folder for files----\\n\")\n",
        "  ! ls -la /opt\n"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "ebXfvQ1qiQHz"
      },
      "source": [
        "# 1.1 Function to set environment\n",
        "def set_spark_env():\n",
        "  print(\"\\n---2. Set Environment variables----\\n\")\n",
        "  os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-8-openjdk-amd64\" \n",
        "  os.environ[\"JRE_HOME\"] = \"/usr/lib/jvm/java-8-openjdk-amd64/jre\" \n",
        "  os.environ[\"SPARK_HOME\"] = \"/opt/spark-3.3.2-bin-hadoop3\" \n",
        "  os.environ[\"SPARK_CONF_DIR\"] = \"/opt/spark-3.3.2-bin-hadoop3/conf\"     \n",
        "  os.environ[\"LD_LIBRARY_PATH\"] += \":/opt/spark-3.3.2-bin-hadoop3/lib/native\"\n",
        "  os.environ[\"PATH\"] += \":/opt/spark-3.3.2-bin-hadoop3/bin:/opt/spark-3.3.2-bin-hadoop3/sbin\"\n",
        "  print(\"\\n---2.1. Check Environment variables----\\n\")\n",
        "  # Check\n",
        "  ! echo $PATH\n",
        "  ! echo $LD_LIBRARY_PATH"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "qUsZggDHj13U"
      },
      "source": [
        "# 1.2 Function to configure spark \n",
        "def spark_conf():\n",
        "  print(\"\\n---3. Configure spark to access hadoop----\\n\")\n",
        "  !mv /opt/spark-3.3.2-bin-hadoop3/conf/spark-env.sh.template  /opt/spark-3.3.2-bin-hadoop3/conf/spark-env.sh\n",
        "  #!echo \"HADOOP_CONF_DIR=/opt/hadoop-3.2.2/etc/hadoop/\" >> /opt/spark-3.1.2-bin-hadoop3.2/conf/spark-env.sh\n",
        "  print(\"\\n---3.1 Check ----\\n\")\n",
        "  #!cat /opt/spark-3.1.1-bin-hadoop3.2/conf/spark-env.sh"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "M7kLbAFLszN3"
      },
      "source": [
        "### Install spark\n"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "z_eaLhtPktHJ"
      },
      "source": [
        "# 2.0 Call all the three functions\n",
        "def install_spark():\n",
        "  spark_koalas_install()\n",
        "  set_spark_env()\n",
        "  spark_conf()\n"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "emaHs1XxRt5z",
        "outputId": "f6d858fd-cc84-4c08-8448-92ef4e3db426",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "source": [
        "# 2.1 \n",
        "install_spark()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "\n",
            "--1.1 Install findspark----\n",
            "\n",
            "\n",
            "--1.3 Download Apache tar.gz----\n",
            "\n",
            "--2023-04-04 05:13:09--  https://dlcdn.apache.org/spark/spark-3.3.2/spark-3.3.2-bin-hadoop3.tgz\n",
            "Resolving dlcdn.apache.org (dlcdn.apache.org)... 151.101.2.132, 2a04:4e42::644\n",
            "Connecting to dlcdn.apache.org (dlcdn.apache.org)|151.101.2.132|:443... connected.\n",
            "HTTP request sent, awaiting response... 200 OK\n",
            "Length: 299360284 (285M) [application/x-gzip]\n",
            "Saving to: ‘spark-3.3.2-bin-hadoop3.tgz’\n",
            "\n",
            "spark-3.3.2-bin-had 100%[===================>] 285.49M   262MB/s    in 1.1s    \n",
            "\n",
            "2023-04-04 05:13:11 (262 MB/s) - ‘spark-3.3.2-bin-hadoop3.tgz’ saved [299360284/299360284]\n",
            "\n",
            "\n",
            "--1.4 Transfer downloaded content and unzip tar.gz----\n",
            "\n",
            "\n",
            "--1.5 Check folder for files----\n",
            "\n",
            "total 292368\n",
            "drwxr-xr-x  1 root root      4096 Apr  4 05:13 .\n",
            "drwxr-xr-x  1 root root      4096 Apr  4 05:11 ..\n",
            "drwxr-xr-x  1 root root      4096 Mar 31 13:52 google\n",
            "drwxr-xr-x  1 root root      4096 Mar 31 13:28 nvidia\n",
            "drwxr-xr-x 13  501 1000      4096 Feb 10 20:40 spark-3.3.2-bin-hadoop3\n",
            "-rw-r--r--  1 root root 299360284 Feb 10 21:28 spark-3.3.2-bin-hadoop3.tgz\n",
            "\n",
            "---2. Set Environment variables----\n",
            "\n",
            "\n",
            "---2.1. Check Environment variables----\n",
            "\n",
            "/opt/bin:/usr/local/nvidia/bin:/usr/local/cuda/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/tools/node/bin:/tools/google-cloud-sdk/bin:/opt/spark-3.3.2-bin-hadoop3/bin:/opt/spark-3.3.2-bin-hadoop3/sbin\n",
            "/usr/local/nvidia/lib:/usr/local/nvidia/lib64:/opt/spark-3.3.2-bin-hadoop3/lib/native\n",
            "\n",
            "---3. Configure spark to access hadoop----\n",
            "\n",
            "\n",
            "---3.1 Check ----\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "MYRPnQhTRdOn"
      },
      "source": [
        "# B. Call libraries\n",
        "We do not import `pandas` but `pyspark.pandas` as `ps`."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "AiO4shcjRjNO",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "54a34e5a-eb7a-4822-f889-2878028bb9d4"
      },
      "source": [
        "# 3.0 Just call some libraries to test\n",
        "import numpy as np\n",
        "import os\n",
        "import time \n",
        "\n",
        "# 3.1 Get spark in sys.path\n",
        "import findspark\n",
        "findspark.init()\n",
        "\n",
        "# 3.2 Call other spark libraries\n",
        "#     Just to test\n",
        "import pyspark.pandas as ps\n",
        "from pyspark.sql import SparkSession\n",
        "#import databricks.koalas as ks"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stderr",
          "text": [
            "/opt/spark-3.3.2-bin-hadoop3/python/pyspark/pandas/__init__.py:49: UserWarning: 'PYARROW_IGNORE_TIMEZONE' environment variable was not set. It is required to set this environment variable to '1' in both driver and executor sides if you use pyarrow>=2.0.0. pandas-on-Spark will set it for you but it does not work if there is a Spark context already launched.\n",
            "  warnings.warn(\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "i7lb_6xCwrZm"
      },
      "source": [
        "# 3.3\n",
        "from IPython.core.interactiveshell import InteractiveShell\n",
        "InteractiveShell.ast_node_interactivity = \"all\""
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "VCOPgKGO60_I",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 37
        },
        "outputId": "72d406e6-65b9-4ade-ea92-db47934e415f"
      },
      "source": [
        "# 3.4 Increase cell width to display wide columnar output\n",
        "from IPython.core.display import display, HTML\n",
        "display(HTML(\"<style>.container { width:100% !important; }</style>\"))"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "display_data",
          "data": {
            "text/plain": [
              "<IPython.core.display.HTML object>"
            ],
            "text/html": [
              "<style>.container { width:100% !important; }</style>"
            ]
          },
          "metadata": {}
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "iHd3YFPHO1SV"
      },
      "source": [
        "# C. Build spark session\n",
        "You can modify spark driver/executor memory here<br>\n",
        "`SparkSession` name is `spark`<br>\n",
        "\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "k5Kb-o82T0oQ"
      },
      "source": [
        "## Modifying spark configuraion\n",
        "Increase driver and executor memory"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "3M6d5xfNJSwx",
        "outputId": "3dff218a-04db-4d91-9ad2-fac3f3fb6e44",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "source": [
        "# 4.0 Check template file\n",
        "! cat /opt/spark-3.3.2-bin-hadoop3/conf/spark-defaults.conf.template"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "#\n",
            "# Licensed to the Apache Software Foundation (ASF) under one or more\n",
            "# contributor license agreements.  See the NOTICE file distributed with\n",
            "# this work for additional information regarding copyright ownership.\n",
            "# The ASF licenses this file to You under the Apache License, Version 2.0\n",
            "# (the \"License\"); you may not use this file except in compliance with\n",
            "# the License.  You may obtain a copy of the License at\n",
            "#\n",
            "#    http://www.apache.org/licenses/LICENSE-2.0\n",
            "#\n",
            "# Unless required by applicable law or agreed to in writing, software\n",
            "# distributed under the License is distributed on an \"AS IS\" BASIS,\n",
            "# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n",
            "# See the License for the specific language governing permissions and\n",
            "# limitations under the License.\n",
            "#\n",
            "\n",
            "# Default system properties included when running spark-submit.\n",
            "# This is useful for setting default environmental settings.\n",
            "\n",
            "# Example:\n",
            "# spark.master                     spark://master:7077\n",
            "# spark.eventLog.enabled           true\n",
            "# spark.eventLog.dir               hdfs://namenode:8021/directory\n",
            "# spark.serializer                 org.apache.spark.serializer.KryoSerializer\n",
            "# spark.driver.memory              5g\n",
            "# spark.executor.extraJavaOptions  -XX:+PrintGCDetails -Dkey=value -Dnumbers=\"one two three\"\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "SXEkJxzyNYn4"
      },
      "source": [
        "# 4.1 Create spark-defaults.conf \n",
        "! cp /opt/spark-3.3.2-bin-hadoop3/conf/spark-defaults.conf.template  /opt/spark-3.3.2-bin-hadoop3/conf/spark-defaults.conf\n"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "oZD3i2MXNacY"
      },
      "source": [
        "# 4.2 Amend properties\n",
        "! echo \"spark.driver.memory 6g\" >> /opt/spark-3.3.2-bin-hadoop3/conf/spark-defaults.conf\n",
        "! echo \"spark.executor.memory 3g\" >> /opt/spark-3.3.2-bin-hadoop3/conf/spark-defaults.conf"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "JtFbJ5UEObIg",
        "outputId": "2c084c17-ef65-4898-ecf8-9686566d2f70",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "source": [
        "# 4.3 Check now\n",
        "! cat /opt/spark-3.3.2-bin-hadoop3/conf/spark-defaults.conf"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "#\n",
            "# Licensed to the Apache Software Foundation (ASF) under one or more\n",
            "# contributor license agreements.  See the NOTICE file distributed with\n",
            "# this work for additional information regarding copyright ownership.\n",
            "# The ASF licenses this file to You under the Apache License, Version 2.0\n",
            "# (the \"License\"); you may not use this file except in compliance with\n",
            "# the License.  You may obtain a copy of the License at\n",
            "#\n",
            "#    http://www.apache.org/licenses/LICENSE-2.0\n",
            "#\n",
            "# Unless required by applicable law or agreed to in writing, software\n",
            "# distributed under the License is distributed on an \"AS IS\" BASIS,\n",
            "# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n",
            "# See the License for the specific language governing permissions and\n",
            "# limitations under the License.\n",
            "#\n",
            "\n",
            "# Default system properties included when running spark-submit.\n",
            "# This is useful for setting default environmental settings.\n",
            "\n",
            "# Example:\n",
            "# spark.master                     spark://master:7077\n",
            "# spark.eventLog.enabled           true\n",
            "# spark.eventLog.dir               hdfs://namenode:8021/directory\n",
            "# spark.serializer                 org.apache.spark.serializer.KryoSerializer\n",
            "# spark.driver.memory              5g\n",
            "# spark.executor.extraJavaOptions  -XX:+PrintGCDetails -Dkey=value -Dnumbers=\"one two three\"\n",
            "spark.driver.memory 6g\n",
            "spark.executor.memory 3g\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Obzrn0jTVslw"
      },
      "source": [
        "## Stop and start SparkSession"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "JDGr6SWsOqox"
      },
      "source": [
        "# 5.0 Build spark session:\n",
        "# Stop spark, if started\n",
        "#spark.stop()\n",
        "# 5.1 Now start spark\n",
        "spark = SparkSession. \\\n",
        "                    builder. \\\n",
        "                    master(\"local[*]\"). \\\n",
        "                    appName(\"myexpt\"). \\\n",
        "                    getOrCreate()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "dFOjGwWiaNZH",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "665552ea-3c22-4eda-b994-222bcae3d33d"
      },
      "source": [
        "sc = spark.sparkContext\n",
        "spark.sparkContext.getConf().getAll()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "[('spark.driver.extraJavaOptions',\n",
              "  '-XX:+IgnoreUnrecognizedVMOptions --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.lang.invoke=ALL-UNNAMED --add-opens=java.base/java.lang.reflect=ALL-UNNAMED --add-opens=java.base/java.io=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.base/java.util.concurrent=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED --add-opens=java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/sun.nio.cs=ALL-UNNAMED --add-opens=java.base/sun.security.action=ALL-UNNAMED --add-opens=java.base/sun.util.calendar=ALL-UNNAMED --add-opens=java.security.jgss/sun.security.krb5=ALL-UNNAMED'),\n",
              " ('spark.executor.memory', '3g'),\n",
              " ('spark.app.name', 'myexpt'),\n",
              " ('spark.app.startTime', '1680585214360'),\n",
              " ('spark.driver.port', '46519'),\n",
              " ('spark.executor.id', 'driver'),\n",
              " ('spark.app.submitTime', '1680585214134'),\n",
              " ('spark.driver.memory', '6g'),\n",
              " ('spark.rdd.compress', 'True'),\n",
              " ('spark.driver.host', '640341b500da'),\n",
              " ('spark.executor.extraJavaOptions',\n",
              "  '-XX:+IgnoreUnrecognizedVMOptions --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.lang.invoke=ALL-UNNAMED --add-opens=java.base/java.lang.reflect=ALL-UNNAMED --add-opens=java.base/java.io=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.base/java.util.concurrent=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED --add-opens=java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/sun.nio.cs=ALL-UNNAMED --add-opens=java.base/sun.security.action=ALL-UNNAMED --add-opens=java.base/sun.util.calendar=ALL-UNNAMED --add-opens=java.security.jgss/sun.security.krb5=ALL-UNNAMED'),\n",
              " ('spark.serializer.objectStreamReset', '100'),\n",
              " ('spark.master', 'local[*]'),\n",
              " ('spark.submit.pyFiles', ''),\n",
              " ('spark.submit.deployMode', 'client'),\n",
              " ('spark.ui.showConsoleProgress', 'true'),\n",
              " ('spark.app.id', 'local-1680585215917')]"
            ]
          },
          "metadata": {},
          "execution_count": 21
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "jimCCywmPJmm",
        "outputId": "0053cce9-5355-476c-bc0b-31e083f0d524"
      },
      "source": [
        "# 5.2\n",
        "print(spark.sparkContext._conf.getAll())"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "[('spark.driver.extraJavaOptions', '-XX:+IgnoreUnrecognizedVMOptions --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.lang.invoke=ALL-UNNAMED --add-opens=java.base/java.lang.reflect=ALL-UNNAMED --add-opens=java.base/java.io=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.base/java.util.concurrent=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED --add-opens=java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/sun.nio.cs=ALL-UNNAMED --add-opens=java.base/sun.security.action=ALL-UNNAMED --add-opens=java.base/sun.util.calendar=ALL-UNNAMED --add-opens=java.security.jgss/sun.security.krb5=ALL-UNNAMED'), ('spark.executor.memory', '3g'), ('spark.app.name', 'myexpt'), ('spark.app.startTime', '1680585214360'), ('spark.driver.port', '46519'), ('spark.executor.id', 'driver'), ('spark.app.submitTime', '1680585214134'), ('spark.driver.memory', '6g'), ('spark.rdd.compress', 'True'), ('spark.driver.host', '640341b500da'), ('spark.executor.extraJavaOptions', '-XX:+IgnoreUnrecognizedVMOptions --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.lang.invoke=ALL-UNNAMED --add-opens=java.base/java.lang.reflect=ALL-UNNAMED --add-opens=java.base/java.io=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.base/java.util.concurrent=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED --add-opens=java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/sun.nio.cs=ALL-UNNAMED --add-opens=java.base/sun.security.action=ALL-UNNAMED --add-opens=java.base/sun.util.calendar=ALL-UNNAMED --add-opens=java.security.jgss/sun.security.krb5=ALL-UNNAMED'), ('spark.serializer.objectStreamReset', '100'), ('spark.master', 'local[*]'), ('spark.submit.pyFiles', ''), ('spark.submit.deployMode', 'client'), ('spark.ui.showConsoleProgress', 'true'), ('spark.app.id', 'local-1680585215917')]\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "L6XYg1IGs17n"
      },
      "source": [
        "# D. Test spark\n",
        "Use existing *spark* session to test if spark is working\n"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "oVlRiGQJmk58",
        "outputId": "d1359dd1-a02a-41ce-fd90-ea477b0829ef",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 320
        }
      },
      "source": [
        "# 6.0 pyspark pandas DataFrame\n",
        "pdf = ps.DataFrame({\n",
        "        'x1': ['a','a','b','b','', 'c', 'd','d'],\n",
        "        'x2': ['apple','', 'orange','orange', 'peach','','apple','orange'],\n",
        "        'x3': [1, 1, 2, 2, 2, 4, 1, 2],\n",
        "        'x4': [2.4, 2.5, 3.5, 1.4, 2.1,1.5, 3.0, 2.0],\n",
        "        'y1': [1, 0, 1, 0, 0, 1, 1, 0],\n",
        "        'y2': ['yes', 'no', 'no','','', 'yes','', 'yes']\n",
        "    })\n",
        "\n",
        "# 6.1\n",
        "pdf"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "  x1      x2  x3   x4  y1   y2\n",
              "0  a   apple   1  2.4   1  yes\n",
              "1  a           1  2.5   0   no\n",
              "2  b  orange   2  3.5   1   no\n",
              "3  b  orange   2  1.4   0     \n",
              "4      peach   2  2.1   0     \n",
              "5  c           4  1.5   1  yes\n",
              "6  d   apple   1  3.0   1     \n",
              "7  d  orange   2  2.0   0  yes"
            ],
            "text/html": [
              "<div>\n",
              "<style scoped>\n",
              "    .dataframe tbody tr th:only-of-type {\n",
              "        vertical-align: middle;\n",
              "    }\n",
              "\n",
              "    .dataframe tbody tr th {\n",
              "        vertical-align: top;\n",
              "    }\n",
              "\n",
              "    .dataframe thead th {\n",
              "        text-align: right;\n",
              "    }\n",
              "</style>\n",
              "<table border=\"1\" class=\"dataframe\">\n",
              "  <thead>\n",
              "    <tr style=\"text-align: right;\">\n",
              "      <th></th>\n",
              "      <th>x1</th>\n",
              "      <th>x2</th>\n",
              "      <th>x3</th>\n",
              "      <th>x4</th>\n",
              "      <th>y1</th>\n",
              "      <th>y2</th>\n",
              "    </tr>\n",
              "  </thead>\n",
              "  <tbody>\n",
              "    <tr>\n",
              "      <th>0</th>\n",
              "      <td>a</td>\n",
              "      <td>apple</td>\n",
              "      <td>1</td>\n",
              "      <td>2.4</td>\n",
              "      <td>1</td>\n",
              "      <td>yes</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>1</th>\n",
              "      <td>a</td>\n",
              "      <td></td>\n",
              "      <td>1</td>\n",
              "      <td>2.5</td>\n",
              "      <td>0</td>\n",
              "      <td>no</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>2</th>\n",
              "      <td>b</td>\n",
              "      <td>orange</td>\n",
              "      <td>2</td>\n",
              "      <td>3.5</td>\n",
              "      <td>1</td>\n",
              "      <td>no</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>3</th>\n",
              "      <td>b</td>\n",
              "      <td>orange</td>\n",
              "      <td>2</td>\n",
              "      <td>1.4</td>\n",
              "      <td>0</td>\n",
              "      <td></td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>4</th>\n",
              "      <td></td>\n",
              "      <td>peach</td>\n",
              "      <td>2</td>\n",
              "      <td>2.1</td>\n",
              "      <td>0</td>\n",
              "      <td></td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>5</th>\n",
              "      <td>c</td>\n",
              "      <td></td>\n",
              "      <td>4</td>\n",
              "      <td>1.5</td>\n",
              "      <td>1</td>\n",
              "      <td>yes</td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>6</th>\n",
              "      <td>d</td>\n",
              "      <td>apple</td>\n",
              "      <td>1</td>\n",
              "      <td>3.0</td>\n",
              "      <td>1</td>\n",
              "      <td></td>\n",
              "    </tr>\n",
              "    <tr>\n",
              "      <th>7</th>\n",
              "      <td>d</td>\n",
              "      <td>orange</td>\n",
              "      <td>2</td>\n",
              "      <td>2.0</td>\n",
              "      <td>0</td>\n",
              "      <td>yes</td>\n",
              "    </tr>\n",
              "  </tbody>\n",
              "</table>\n",
              "</div>"
            ]
          },
          "metadata": {},
          "execution_count": 23
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "L3pSQ7sDmnjt",
        "outputId": "11b548c0-8dfa-45bc-834c-f1254388a138",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "source": [
        "# 6.2 Transform to Spark DataFrame\n",
        "#     and print\n",
        "df = pdf.to_spark()\n",
        "df.show()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stderr",
          "text": [
            "/opt/spark-3.3.2-bin-hadoop3/python/pyspark/pandas/utils.py:975: PandasAPIOnSparkAdviceWarning: If `index_col` is not specified for `to_spark`, the existing index is lost when converting to Spark DataFrame.\n",
            "  warnings.warn(message, PandasAPIOnSparkAdviceWarning)\n"
          ]
        },
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+---+------+---+---+---+---+\n",
            "| x1|    x2| x3| x4| y1| y2|\n",
            "+---+------+---+---+---+---+\n",
            "|  a| apple|  1|2.4|  1|yes|\n",
            "|  a|      |  1|2.5|  0| no|\n",
            "|  b|orange|  2|3.5|  1| no|\n",
            "|  b|orange|  2|1.4|  0|   |\n",
            "|   | peach|  2|2.1|  0|   |\n",
            "|  c|      |  4|1.5|  1|yes|\n",
            "|  d| apple|  1|3.0|  1|   |\n",
            "|  d|orange|  2|2.0|  0|yes|\n",
            "+---+------+---+---+---+---+\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "sTjT6sQYbgz3"
      },
      "source": [
        "############"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "QxLQzJD4p81p"
      },
      "source": [
        "# E. Some useful functions\n",
        "Execute only if you need. Else, forget it"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "pYwZ4OJfp_tp"
      },
      "source": [
        "# 7.0 Per column how many null values:\n",
        "from pyspark.sql.functions import isnan, when, count, col\n",
        "def null_values(data):\n",
        "  data.select([count(when(isnan(c) | col(c).isNull(), c)).alias(c) for c in data.columns]).show()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "W8GzRFOIUKoj"
      },
      "source": [
        "# 8.0 Finding mode of a column\n",
        "# Refer StackOverflow: https://stackoverflow.com/a/58279672\n",
        "def mode(df,col):\n",
        "  df.groupby(\"col\").count().orderBy(\"count\", ascending=False).first()[0]\n",
        "\n",
        "# 9.0 Find mode of all columns\n",
        "def mode_cols(df):\n",
        "  [[i,df.groupby(i).count().orderBy(\"count\", ascending=False).first()[0]] for i in df.columns]  "
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "IbIOI7qpebYo"
      },
      "source": [
        "# 10.0 Value counts\n",
        "def value_counts(df):\n",
        "    for colm in df.columns:\n",
        "        df.groupby(colm).count().show()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "PQYbaYIBW-BS"
      },
      "source": [
        "# 11.0 Map a string to another\n",
        "# See here: https://stackoverflow.com/a/55026324/3282777\n",
        "# Code not tested\n",
        "# Map a column 'colname' in dataframe 'df'\n",
        "# as follows:\n",
        "#\n",
        "# map_dict= {\n",
        "#            'A': '1',\n",
        "#            'B': '2'\n",
        "#           }\n",
        "\n",
        "def mapping(df,map_dict, colname):\n",
        "  df2 = df.replace(to_replace=map_dict, subset=['yourColName'])"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Zn3xhm6LdXvK"
      },
      "source": [
        "# F. Your experiments\n",
        "SparkSession is <i>'spark'</i>. Call all needed libraries. Begin with mounting gdrive."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "VQ2e6elQoj5z",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "488e9d72-9a30-48bc-a4a3-2ca9339162e7"
      },
      "source": [
        "# 1.0 Mount gdrive\n",
        "from google.colab import drive\n",
        "drive.mount('/gdrive')"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Mounted at /gdrive\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# 2.0 Call ML libraries:\n",
        "\n",
        "from pyspark.ml.feature import StringIndexer, VectorAssembler, OneHotEncoder\n",
        "from pyspark.ml.classification import RandomForestClassifier\n",
        "from pyspark.ml.evaluation import BinaryClassificationEvaluator\n",
        "from pyspark.ml import Pipeline"
      ],
      "metadata": {
        "id": "YbLsLxAlGkX-"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# 2.1 List path to folder containg csv file. \n",
        "#      Note '/' at the end:\n",
        "\n",
        "pathToFolder = \"/gdrive/MyDrive/Colab_data_files/bankmarketing/\""
      ],
      "metadata": {
        "id": "RcIdkz42UAJA"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# 2.2 Read data:\n",
        "\n",
        "df = ps.read_csv(\n",
        "                 pathToFolder + \"bank-full.csv\",\n",
        "                 sep= \";\"\n",
        "                 )\n"
      ],
      "metadata": {
        "id": "busGA6hBKzvb",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "327bd329-3145-4aa7-9b54-2361a0c0a06f"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stderr",
          "text": [
            "/opt/spark-3.3.2-bin-hadoop3/python/pyspark/pandas/utils.py:975: PandasAPIOnSparkAdviceWarning: If `index_col` is not specified for `read_csv`, the default index is attached which can cause additional overhead.\n",
            "  warnings.warn(message, PandasAPIOnSparkAdviceWarning)\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# 2.3 Your cat cols and num cols:\n",
        "\n",
        "catCols = df.select_dtypes(include = ['object']).columns\n",
        "numCols = df.select_dtypes(include = ['int32']).columns\n",
        "\n",
        "catCols = catCols.to_list()\n",
        "print(\"\\n============\\n\")\n",
        "numCols = numCols.to_list()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "fPCywHuTLbc6",
        "outputId": "c3c3349f-9467-4415-c839-4c958b655e26"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "\n",
            "============\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# 2.4 What names catCols will have after StringIndexing and OHE:\n",
        "\n",
        "outCols = [i + \"_index\" for i in list(catCols)[:-1]]  # Exclude y\n",
        "outCols\n",
        "oheCols = [i + \"_vec\" for i in list(catCols)[:-1]]\n",
        "oheCols\n",
        "\n",
        "vaCols = oheCols.copy()\n",
        "vaCols.extend(numCols)\n",
        "vaCols"
      ],
      "metadata": {
        "id": "ExC0ftYSLO7N"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "For PCA, see [here](https://spark.apache.org/docs/latest/ml-features.html#pca)"
      ],
      "metadata": {
        "id": "tJuP1wl8sTeB"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "from pyspark.ml.feature import PCA"
      ],
      "metadata": {
        "id": "dM3BaCIisb3u"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# 3.0 Instantiate StringIndexer for multiple columns:\n",
        "\n",
        "indexer = StringIndexer(\n",
        "                        inputCols=catCols[:-1],   # Exclude y\n",
        "                        outputCols=outCols\n",
        "                        )\n",
        "\n",
        "\n",
        "encoder = OneHotEncoder(inputCols=outCols,\n",
        "                        outputCols=oheCols)\n",
        "\n",
        "\n",
        "yindex = StringIndexer(inputCol = 'y',\n",
        "                       outputCol = 'y_index')\n",
        "\n",
        "\n",
        "vectorAssembler = VectorAssembler(\n",
        "                                   inputCols = vaCols,\n",
        "                                   outputCol = 'v_features'\n",
        "                                  )\n",
        "\n",
        "pca =  PCA(k=3, inputCol=\"v_features\", outputCol=\"features\")\n",
        "\n",
        "rf = RandomForestClassifier(\n",
        "                             labelCol = 'y_index',\n",
        "                             numTrees=100\n",
        "                            )"
      ],
      "metadata": {
        "id": "bei-YA22M1Mb"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# 3.1 Split spark_df into train/test:\n",
        "\n",
        "splits = df.to_spark().randomSplit([0.7, 0.3])\n",
        "type(splits)   # list\n"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "947eff0c-eec0-44ca-df67-0770820251d4",
        "id": "o9vH0RNRQDcX"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stderr",
          "text": [
            "/opt/spark-3.3.2-bin-hadoop3/python/pyspark/pandas/utils.py:975: PandasAPIOnSparkAdviceWarning: If `index_col` is not specified for `to_spark`, the existing index is lost when converting to Spark DataFrame.\n",
            "  warnings.warn(message, PandasAPIOnSparkAdviceWarning)\n"
          ]
        },
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "list"
            ]
          },
          "metadata": {},
          "execution_count": 34
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# 3.2\n",
        "train_df = splits[0]\n",
        "test_df = splits[1]"
      ],
      "metadata": {
        "id": "YcvBu8ESQDcb"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "## What are pipelines:\n",
        "See [here](https://spark.apache.org/docs/latest/ml-pipeline.html)"
      ],
      "metadata": {
        "id": "-HxUn6kEpiTf"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "# 4.0 Pipeline now:\n",
        "pipeline = Pipeline(stages= [indexer, encoder, vectorAssembler, yindex, pca, rf ])"
      ],
      "metadata": {
        "id": "C-5E8eKYIWo0"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# 4.1 fit pipeline:\n",
        "model_p = pipeline.fit(train_df)"
      ],
      "metadata": {
        "id": "HbapVJSBJFte"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# 5.0 Make predictions:\n",
        "\n",
        "test_df = model_p.transform(test_df)"
      ],
      "metadata": {
        "id": "pNfS4YO-KjB_"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# 5.1:\n",
        "\n",
        "test_df.show(2)"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "WaMuCohhKtJ7",
        "outputId": "9a8ae848-8926-4f55-f463-059c5af28158"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+---+-------+-------+---------+------------------+----------------+--------------+---------------+---------------+---------------------+----------------------+---------------+-----------------------------------+-----+---------------------------+-----------------------------+---+---------+-------------+---------------+------------------------+--------------------+---------------------+---------------------+----------------------------+-----------------------------------+---------------+-------------+-------------+----------------------+------------------+-------------------+-------------------+--------------------------+---------------------------------+--------------------+-------+--------------------+--------------------+--------------------+----------+\n",
            "|age|    job|marital|education|hascreditindefault|avgyearlybalance|hashousingloan|haspersonalloan|contactCommType|lastContactdayOfMonth|lastContactmonthOfYear|contactDuration|noOfContactsPerformedDuringcampaign|pdays|noOfContprevious_tocampaign|outcomeOfPreMarketingcampaign|  y|job_index|marital_index|education_index|hascreditindefault_index|hashousingloan_index|haspersonalloan_index|contactCommType_index|lastContactmonthOfYear_index|outcomeOfPreMarketingcampaign_index|        job_vec|  marital_vec|education_vec|hascreditindefault_vec|hashousingloan_vec|haspersonalloan_vec|contactCommType_vec|lastContactmonthOfYear_vec|outcomeOfPreMarketingcampaign_vec|          v_features|y_index|            features|       rawPrediction|         probability|prediction|\n",
            "+---+-------+-------+---------+------------------+----------------+--------------+---------------+---------------+---------------------+----------------------+---------------+-----------------------------------+-----+---------------------------+-----------------------------+---+---------+-------------+---------------+------------------------+--------------------+---------------------+---------------------+----------------------------+-----------------------------------+---------------+-------------+-------------+----------------------+------------------+-------------------+-------------------+--------------------------+---------------------------------+--------------------+-------+--------------------+--------------------+--------------------+----------+\n",
            "| 18|student| single|  unknown|                no|               3|            no|             no|       cellular|                   25|                   aug|            130|                                  2|   -1|                          0|                      unknown|yes|     10.0|          1.0|            3.0|                     0.0|                 1.0|                  0.0|                  0.0|                         2.0|                                0.0|(11,[10],[1.0])|(2,[1],[1.0])|    (3,[],[])|         (1,[0],[1.0])|         (1,[],[])|      (1,[0],[1.0])|      (2,[0],[1.0])|            (11,[2],[1.0])|                    (3,[0],[1.0])|(42,[10,12,16,18,...|    1.0|[3.22817781567712...|[94.7912037566771...|[0.94791203756677...|       0.0|\n",
            "| 19|student| single|  primary|                no|              56|            no|             no|       cellular|                   12|                   aug|            246|                                  1|   -1|                          0|                      unknown| no|     10.0|          1.0|            2.0|                     0.0|                 1.0|                  0.0|                  0.0|                         2.0|                                0.0|(11,[10],[1.0])|(2,[1],[1.0])|(3,[2],[1.0])|         (1,[0],[1.0])|         (1,[],[])|      (1,[0],[1.0])|      (2,[0],[1.0])|            (11,[2],[1.0])|                    (3,[0],[1.0])|(42,[10,12,15,16,...|    0.0|[56.4261615001661...|[92.654753319275,...|[0.92654753319274...|       0.0|\n",
            "+---+-------+-------+---------+------------------+----------------+--------------+---------------+---------------+---------------------+----------------------+---------------+-----------------------------------+-----+---------------------------+-----------------------------+---+---------+-------------+---------------+------------------------+--------------------+---------------------+---------------------+----------------------------+-----------------------------------+---------------+-------------+-------------+----------------------+------------------+-------------------+-------------------+--------------------------+---------------------------------+--------------------+-------+--------------------+--------------------+--------------------+----------+\n",
            "only showing top 2 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# 5.2 Evaluate:\n",
        "\n",
        "evaluator = BinaryClassificationEvaluator(labelCol = 'y_index', metricName = 'areaUnderROC' )"
      ],
      "metadata": {
        "id": "iZ1HPiZgQdjz"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# 5.2.1\n",
        "evaluator.evaluate(test_df)   # 0.836281717074084"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "e5981fb2-5871-46fc-dab9-48ca230c4c18",
        "id": "D9s1UqtAQdj0"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "0.8362817170740842"
            ]
          },
          "metadata": {},
          "execution_count": 41
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "## Two exercises:\n",
        "### Exercise 1:\n",
        "#    Before PCA, perform Standard Sclaing also.\n",
        "#       See: https://spark.apache.org/docs/latest/ml-features.html#standardscaler\n",
        "#    What about using imputer? Imputer does not impute strings. First index and then impute.\n",
        "#       https://spark.apache.org/docs/latest/ml-features.html#imputer\n",
        "#       https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.feature.Imputer.html"
      ],
      "metadata": {
        "id": "FNFZefZ-tP6e"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "### Exercise 2:\n",
        "# Dataset adultdata_modified.csv at this path (\"/gdrive/MyDrive/Colab_data_files/adult/\") in gdrive\n",
        "#  has Null values. Develop a pipelining model to perform classification."
      ],
      "metadata": {
        "id": "40-1W-dd8CEh"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "################ I am done #################3"
      ],
      "metadata": {
        "id": "GPU3rQKspCqp"
      },
      "execution_count": null,
      "outputs": []
    }
  ]
}