Recommendation_with_Pyspark+SystemDS.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "SystemDS on Colaboratory.ipynb",
      "provenance": [],
      "collapsed_sections": [],
      "toc_visible": true,
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/j143/notebooks/blob/master/Recommendation_with_Pyspark%2BSystemDS.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "XX60cA7YuZsw",
        "colab_type": "text"
      },
      "source": [
        "##### Copyright &copy; 2020 The Apache Software Foundation."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "8GEGDZ9GuZGp",
        "colab_type": "code",
        "cellView": "form",
        "colab": {}
      },
      "source": [
        "# @title Apache Version 2.0 (The \"License\");\n",
        "#-------------------------------------------------------------\n",
        "#\n",
        "# Licensed to the Apache Software Foundation (ASF) under one\n",
        "# or more contributor license agreements.  See the NOTICE file\n",
        "# distributed with this work for additional information\n",
        "# regarding copyright ownership.  The ASF licenses this file\n",
        "# to you under the Apache License, Version 2.0 (the\n",
        "# \"License\"); you may not use this file except in compliance\n",
        "# with 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,\n",
        "# software distributed under the License is distributed on an\n",
        "# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n",
        "# KIND, either express or implied.  See the License for the\n",
        "# specific language governing permissions and limitations\n",
        "# under the License.\n",
        "#\n",
        "#-------------------------------------------------------------"
      ],
      "execution_count": 1,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "_BbCdLjRoy2A",
        "colab_type": "text"
      },
      "source": [
        "### Developer notebook for Apache SystemDS"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "zhdfvxkEq1BX",
        "colab_type": "text"
      },
      "source": [
        "Run this notebook online at [Google Colab ↗](https://colab.research.google.com/github/apache/systemds/blob/master/notebooks/systemds_dev.ipynb).\n",
        "\n",
        "\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "efFVuggts1hr",
        "colab_type": "text"
      },
      "source": [
        "This Jupyter/Colab-based tutorial will interactively walk through development setup and running SystemDS in both the\n",
        "\n",
        "A. standalone mode \\\n",
        "B. with Apache Spark.\n",
        "\n",
        "Flow of the notebook:\n",
        "1. Download and Install the dependencies\n",
        "2. Go to section **A** or **B**"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "vBC5JPhkGbIV",
        "colab_type": "text"
      },
      "source": [
        "#### Download and Install the dependencies\n",
        "\n",
        "1. **Runtime:** Java (OpenJDK 8 is preferred)\n",
        "2. **Build:** Apache Maven\n",
        "3. **Backend:** Apache Spark (optional)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "VkLasseNylPO",
        "colab_type": "text"
      },
      "source": [
        "##### Setup\n",
        "\n",
        "A custom function to run OS commands."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "4Wmf-7jfydVH",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# Run and print a shell command.\n",
        "def run(command):\n",
        "  print('>> {}'.format(command))\n",
        "  !{command}\n",
        "  print('')"
      ],
      "execution_count": 7,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "kvD4HBMi0ohY",
        "colab_type": "text"
      },
      "source": [
        "##### Install Java\n",
        "Let us install OpenJDK 8. More about [OpenJDK ↗](https://openjdk.java.net/install/)."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "aosNm6oweGqW",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 247
        },
        "outputId": "e43a8ee8-ee89-4aab-d8d3-bfe642a2a685"
      },
      "source": [
        "!apt-get update"
      ],
      "execution_count": 30,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "Get:1 http://security.ubuntu.com/ubuntu bionic-security InRelease [88.7 kB]\n",
            "Ign:2 https://developer.download.nvidia.com/compute/cuda/repos/ubuntu1804/x86_64  InRelease\n",
            "Get:3 https://cloud.r-project.org/bin/linux/ubuntu bionic-cran35/ InRelease [3,626 B]\n",
            "Hit:4 http://ppa.launchpad.net/graphics-drivers/ppa/ubuntu bionic InRelease\n",
            "Ign:5 https://developer.download.nvidia.com/compute/machine-learning/repos/ubuntu1804/x86_64  InRelease\n",
            "Hit:6 https://developer.download.nvidia.com/compute/cuda/repos/ubuntu1804/x86_64  Release\n",
            "Hit:7 http://archive.ubuntu.com/ubuntu bionic InRelease\n",
            "Hit:8 https://developer.download.nvidia.com/compute/machine-learning/repos/ubuntu1804/x86_64  Release\n",
            "Get:9 http://archive.ubuntu.com/ubuntu bionic-updates InRelease [88.7 kB]\n",
            "Hit:10 http://ppa.launchpad.net/marutter/c2d4u3.5/ubuntu bionic InRelease\n",
            "Get:11 http://archive.ubuntu.com/ubuntu bionic-backports InRelease [74.6 kB]\n",
            "Fetched 256 kB in 2s (131 kB/s)\n",
            "Reading package lists...\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "8Xnb_ePUyQIL",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 70
        },
        "outputId": "4b34ac2c-c897-45c8-f133-fb041bfe5946"
      },
      "source": [
        "!apt-get install openjdk-8-jdk-headless -qq > /dev/null\n",
        "\n",
        "# run the below command to replace the existing installation\n",
        "!update-alternatives --set java /usr/lib/jvm/java-8-openjdk-amd64/jre/bin/java\n",
        "\n",
        "import os\n",
        "os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-8-openjdk-amd64\"\n",
        "\n",
        "!java -version"
      ],
      "execution_count": 9,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "openjdk version \"1.8.0_265\"\n",
            "OpenJDK Runtime Environment (build 1.8.0_265-8u265-b01-0ubuntu2~18.04-b01)\n",
            "OpenJDK 64-Bit Server VM (build 25.265-b01, mixed mode)\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Xphbe3R43XLw",
        "colab_type": "text"
      },
      "source": [
        "##### Install Apache Spark (Optional, if you want to work with spark backend)\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "_WgEa00pTs3w",
        "colab_type": "text"
      },
      "source": [
        "NOTE: If spark is not downloaded. Let us make sure the version we are trying to download is officially supported at\n",
        "https://spark.apache.org/downloads.html"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "3zdtkFkLnskx",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# Spark and Hadoop version\n",
        "spark_version = 'spark-2.4.6'\n",
        "hadoop_version = 'hadoop2.7'\n",
        "spark_path = f\"/opt/{spark_version}-bin-{hadoop_version}\"\n",
        "if not os.path.exists(spark_path):\n",
        "  run(f\"wget -q -nc -O apache-spark.tgz https://downloads.apache.org/spark/{spark_version}/{spark_version}-bin-{hadoop_version}.tgz\")\n",
        "  run('tar zxf apache-spark.tgz -C /opt')\n",
        "  run('rm -f apache-spark.tgz')\n",
        "\n",
        "os.environ[\"SPARK_HOME\"] = spark_path\n",
        "os.environ[\"PATH\"] += \":$SPARK_HOME/bin\"\n"
      ],
      "execution_count": 6,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "9ozgLRLrodPI",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "import os\n",
        "\n",
        "os.environ[\"SPARK_HOME\"] = spark_path\n",
        "os.environ[\"PATH\"] += \":$SPARK_HOME/bin\""
      ],
      "execution_count": 7,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "xmDLN3AMkXW3",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 247
        },
        "outputId": "fabe05dd-15dd-45b6-bccc-48fc4520e307"
      },
      "source": [
        "# Download the maven source.\n",
        "maven_version = 'apache-maven-3.6.3'\n",
        "maven_path = f\"/opt/{maven_version}\"\n",
        "\n",
        "if not os.path.exists(maven_path):\n",
        "  run(f\"wget -q -nc -O apache-maven.zip https://downloads.apache.org/maven/maven-3/3.6.3/binaries/{maven_version}-bin.zip\")\n",
        "  run('unzip -q -d /opt apache-maven.zip')\n",
        "  run('rm -f apache-maven.zip')\n",
        "\n",
        "# Let's choose the absolute path instead of $PATH environment variable.\n",
        "def maven(args):\n",
        "  run(f\"{maven_path}/bin/mvn {args}\")\n",
        "\n",
        "maven('-v')"
      ],
      "execution_count": 25,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            ">> wget -q -nc -O apache-maven.zip https://downloads.apache.org/maven/maven-3/3.6.3/binaries/apache-maven-3.6.3-bin.zip\n",
            "\n",
            ">> unzip -q -d /opt apache-maven.zip\n",
            "\n",
            ">> rm -f apache-maven.zip\n",
            "\n",
            ">> /opt/apache-maven-3.6.3/bin/mvn -v\n",
            "\u001b[1mApache Maven 3.6.3 (cecedd343002696d0abb50b32b541b8a6ba2883f)\u001b[m\n",
            "Maven home: /opt/apache-maven-3.6.3\n",
            "Java version: 1.8.0_265, vendor: Private Build, runtime: /usr/lib/jvm/java-8-openjdk-amd64/jre\n",
            "Default locale: en_US, platform encoding: UTF-8\n",
            "OS name: \"linux\", version: \"4.19.112+\", arch: \"amd64\", family: \"unix\"\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "2QWy5l8JfjLa",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "!pip install -q findspark"
      ],
      "execution_count": 3,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "JULdE-xMfqlK",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "import findspark\n",
        "findspark.init()"
      ],
      "execution_count": 8,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "QCr-5vFVhSZ2",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from pyspark import SparkContext, SparkConf\n",
        "\n",
        "conf = SparkConf().setAppName(\"systemds\").setMaster(\"local[*]\")\n",
        "sc = SparkContext(conf=conf)"
      ],
      "execution_count": 9,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "TTPvvBblOS4a",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from pyspark.sql import SparkSession\n",
        "\n",
        "spark = SparkSession(sc)"
      ],
      "execution_count": 10,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "91pJ5U8k3cjk",
        "colab_type": "text"
      },
      "source": [
        "#### Get Apache SystemDS\n",
        "\n",
        "Apache SystemDS development happens on GitHub at [apache/systemds ↗](https://github.com/apache/systemds)"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "SaPIprmg3lKE",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 158
        },
        "outputId": "43b97bbc-3525-494c-ee99-ba6c60d4b4ea"
      },
      "source": [
        "!git clone https://github.com/apache/systemds systemds --depth=1\n",
        "%cd systemds"
      ],
      "execution_count": 10,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "Cloning into 'systemds'...\n",
            "remote: Enumerating objects: 7642, done.\u001b[K\n",
            "remote: Counting objects: 100% (7642/7642), done.\u001b[K\n",
            "remote: Compressing objects: 100% (4588/4588), done.\u001b[K\n",
            "remote: Total 7642 (delta 5595), reused 3643 (delta 2967), pack-reused 0\u001b[K\n",
            "Receiving objects: 100% (7642/7642), 14.79 MiB | 11.47 MiB/s, done.\n",
            "Resolving deltas: 100% (5595/5595), done.\n",
            "/content/systemds\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "bVSkut6ElTnQ",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 34
        },
        "outputId": "a013a222-8842-4dd7-f932-2434b04a1b18"
      },
      "source": [
        "%cd systemds"
      ],
      "execution_count": 31,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "/content/systemds\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "CUzGgqVeknDH",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "!/opt/apache-maven-3.6.3/bin/mvn clean package -P distribution"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Q7-2D-HRmgm5",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 34
        },
        "outputId": "a98cbbab-663b-4030-9e53-0d44be9c8fbd"
      },
      "source": [
        "%cd src/main/python"
      ],
      "execution_count": 33,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "/content/systemds/src/main/python\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "CV0vwHG3m_qq",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "!python pre_setup.py"
      ],
      "execution_count": 35,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "eMQR6Ui9na0w",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "!pip install ."
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "zGOswvv4iWy4",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 70
        },
        "outputId": "b732e5d7-f0e1-4b94-b58c-c8323ff58391"
      },
      "source": [
        "!pip install systemds"
      ],
      "execution_count": 22,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "Requirement already satisfied: systemds in /usr/local/lib/python3.6/dist-packages (0.2.0)\n",
            "Requirement already satisfied: py4j>=0.10.0 in /usr/local/lib/python3.6/dist-packages (from systemds) (0.10.9)\n",
            "Requirement already satisfied: numpy>=1.8.2 in /usr/local/lib/python3.6/dist-packages (from systemds) (1.18.5)\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "K2Tc7GrpTxev",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 70
        },
        "outputId": "e416e51b-5d7c-4444-f96b-f8fda8923d0a"
      },
      "source": [
        "%%sh\n",
        "\n",
        "# Download dataset\n",
        "curl -O http://snap.stanford.edu/data/amazon0601.txt.gz\n",
        "gunzip amazon0601.txt.gz"
      ],
      "execution_count": 11,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current\n",
            "                                 Dload  Upload   Total   Spent    Left  Speed\n",
            "\r  0     0    0     0    0     0      0      0 --:--:-- --:--:-- --:--:--     0\r  1 11.2M    1  203k    0     0   303k      0  0:00:37 --:--:--  0:00:37  303k\r 14 11.2M   14 1683k    0     0   984k      0  0:00:11  0:00:01  0:00:10  983k\r 43 11.2M   43 5002k    0     0  1863k      0  0:00:06  0:00:02  0:00:04 1863k\r 81 11.2M   81 9394k    0     0  2566k      0  0:00:04  0:00:03  0:00:01 2566k\r100 11.2M  100 11.2M    0     0  2825k      0  0:00:04  0:00:04 --:--:-- 2825k\n"
          ],
          "name": "stderr"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "NseiyogsVd61",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 194
        },
        "outputId": "b5655efb-5ec8-4718-df41-04b6379955e8"
      },
      "source": [
        "!sed -n 1,10p amazon0601.txt"
      ],
      "execution_count": 12,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "# Directed graph (each unordered pair of nodes is saved once): Amazon0601.txt \r\n",
            "# Amazon product co-purchaisng network from June 01 2003\r\n",
            "# Nodes: 403394 Edges: 3387388\r\n",
            "# FromNodeId\tToNodeId\r\n",
            "0\t1\r\n",
            "0\t2\r\n",
            "0\t3\r\n",
            "0\t4\r\n",
            "0\t5\r\n",
            "0\t6\r\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "BJc7FoHuXaKb",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "import pyspark.sql.functions as F\n"
      ],
      "execution_count": 11,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "TmzPaTd9OoGK",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 52
        },
        "outputId": "0b6d4e40-07d1-42f3-8e1f-224273baa632"
      },
      "source": [
        "!ls"
      ],
      "execution_count": 19,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "amazon0601.txt\tconf\t\t dev\t docs\t  notebooks  pom.xml\tscripts\n",
            "bin\t\tCONTRIBUTING.md  docker  LICENSE  NOTICE     README.md\tsrc\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "4pwTKb0BXgk1",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "dataPath = \"/content/amazon0601.txt\"\n",
        "\n",
        "X_train = (sc.textFile(dataPath)\n",
        "    .filter(lambda l: not l.startswith(\"#\"))\n",
        "    .map(lambda l: l.split(\"\\t\"))\n",
        "    .map(lambda prods: (int(prods[0]), int(prods[1]), 1.0))\n",
        "    .toDF((\"prod_i\", \"prod_j\", \"x_ij\"))\n",
        "    .filter(\"prod_i < 500 AND prod_j < 500\") # Filter for memory constraints\n",
        "    .cache())"
      ],
      "execution_count": 12,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "YPXMuzq8O4gr",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 34
        },
        "outputId": "4d45a63c-1e25-452f-c4fb-94c65bcc32c0"
      },
      "source": [
        "X_train"
      ],
      "execution_count": 16,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "DataFrame[prod_i: bigint, prod_j: bigint, x_ij: double]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 16
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Xe20U0E7ilUY",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# Maximum product id, is 499\n",
        "max_prod_i = X_train.select(F.max(\"prod_i\")).first()[0]"
      ],
      "execution_count": 13,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "8UQ-fi7yPCSm",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "max_prod_j = X_train.select(F.max(\"prod_j\")).first()[0]"
      ],
      "execution_count": 14,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "dyYBSENaPjVr",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "numProducts = max(max_prod_i, max_prod_j) + 1 # since 0-based indexing"
      ],
      "execution_count": 15,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "jw84vSynPwHs",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 34
        },
        "outputId": "8581745e-17bf-4be8-93f6-a740517d6591"
      },
      "source": [
        "print(\"Total number of products: {}\".format(numProducts))"
      ],
      "execution_count": 16,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "Total number of products: 500\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "J5EftiqxjUn-",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "import systemds\n",
        "from systemds.context import SystemDSContext"
      ],
      "execution_count": 1,
      "outputs": []
    }
  ]
}