notebooks/official/workbench/spark/spark_bigquery.ipynb

{
  "cells": [
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "ur8xi4C7S06n"
      },
      "outputs": [],
      "source": [
        "# Copyright 2022 Google LLC\n",
        "#\n",
        "# Licensed under the Apache License, Version 2.0 (the \"License\");\n",
        "# you may not use this file except in compliance with the License.\n",
        "# You may obtain a copy of the License at\n",
        "#\n",
        "#     https://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."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "JAPoU8Sm5E6e"
      },
      "source": [
        "# Digest and analyze data from BigQuery with Dataproc\n",
        "\n",
        "<table align=\"left\">\n",
        "  <td style=\"text-align: center\">\n",
        "    <a href=\"https://colab.research.google.com/github/GoogleCloudPlatform/vertex-ai-samples/blob/main/notebooks/official/workbench/spark/spark_bigquery.ipynb\">\n",
        "      <img src=\"https://cloud.google.com/ml-engine/images/colab-logo-32px.png\" alt=\"Google Colaboratory logo\"><br> Open in Colab\n",
        "    </a>\n",
        "  </td>\n",
        "  <td style=\"text-align: center\">\n",
        "    <a href=\"https://console.cloud.google.com/vertex-ai/colab/import/https:%2F%2Fraw.githubusercontent.com%2FGoogleCloudPlatform%2Fvertex-ai-samples%2Fmain%2Fnotebooks%2Fofficial%2Fworkbench%2Fspark%2Fspark_bigquery.ipynb\">\n",
        "      <img width=\"32px\" src=\"https://cloud.google.com/ml-engine/images/colab-enterprise-logo-32px.png\" alt=\"Google Cloud Colab Enterprise logo\"><br> Open in Colab Enterprise\n",
        "    </a>\n",
        "  </td>    \n",
        "  <td style=\"text-align: center\">\n",
        "    <a href=\"https://console.cloud.google.com/vertex-ai/workbench/deploy-notebook?download_url=https://raw.githubusercontent.com/GoogleCloudPlatform/vertex-ai-samples/main/notebooks/official/workbench/spark/spark_bigquery.ipynb\">\n",
        "      <img src=\"https://lh3.googleusercontent.com/UiNooY4LUgW_oTvpsNhPpQzsstV5W8F7rYgxgGBD85cWJoLmrOzhVs_ksK_vgx40SHs7jCqkTkCk=e14-rj-sc0xffffff-h130-w32\" alt=\"Vertex AI logo\"><br> Open in Workbench\n",
        "    </a>\n",
        "  </td>\n",
        "  <td style=\"text-align: center\">\n",
        "    <a href=\"https://github.com/GoogleCloudPlatform/vertex-ai-samples/blob/main/notebooks/official/workbench/spark/spark_bigquery.ipynb\">\n",
        "      <img src=\"https://cloud.google.com/ml-engine/images/github-logo-32px.png\" alt=\"GitHub logo\"><br> View on GitHub\n",
        "    </a>\n",
        "  </td>\n",
        "</table>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "tvgnzT1CKxrO"
      },
      "source": [
        "## Overview\n",
        "\n",
        "This notebook tutorial shows you how to ingest, analyze, and write data to BigQuery using Apache Spark with [Dataproc](https://cloud.google.com/dataproc). The notebook code analyzes GitHub Activity Data to explore metrics related to programming languages used in GitHub repositories.\n",
        "\n",
        "To run this notebook, click the link `Open in Vertex AI Workbench` above.\n",
        "\n",
        "Learn more about [Vertex AI Workbench](https://cloud.google.com/vertex-ai/docs/workbench/introduction) and [Dataproc Serverless for Spark](https://cloud.google.com/dataproc-serverless/docs/guides/bigquery-connector-spark-example)."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "tvgnzT1CKxrO"
      },
      "source": [
        "### Objective\n",
        "\n",
        "This notebook tutorial runs an Apache Spark job that fetches data from the BigQuery \"GitHub Activity Data\" dataset, queries the data, and then writes the results back to BigQuery. This job sequence represents a common data engineering use case: ingesting, transforming, and querying data, and then writing the output to a database. It also demonstrates how to submit an Apache Spark job to Dataproc.\n",
        "\n",
        "This tutorial uses the following Google Cloud ML services:\n",
        "\n",
        "- Dataproc\n",
        "- BigQuery\n",
        "\n",
        "The steps performed are:\n",
        "\n",
        "- Setting up a Google Cloud project and Dataproc cluster.\n",
        "- Configuring the spark-bigquery-connector.\n",
        "- Ingesting data from BigQuery into a Spark DataFrame.\n",
        "- Preprocessing ingested data.\n",
        "- Querying the most frequently used programming language in monoglot repos.\n",
        "- Querying the average size (MB) of code in each language stored in monoglot repos.\n",
        "- Querying the languages files most frequently found together in polyglot repos.\n",
        "- Writing the query results back into BigQuery.\n",
        "- Deleting the resources created for this notebook tutorial."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "tvgnzT1CKxrO"
      },
      "source": [
        "### Dataset\n",
        "\n",
        "The [GitHub Activity Data](https://console.cloud.google.com/marketplace/product/github/github-repos) dataset is available in [BigQuery Public Datasets](https://cloud.google.com/bigquery/public-data), and provides free querying of up to 1TB of data each month. It contains data on two different types of repositories: \"polyglot\" repos, which support multiple programming language files, and \"monoglot\" repos, which support one programming language."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "tvgnzT1CKxrO"
      },
      "source": [
        "### Costs \n",
        "\n",
        "This tutorial uses billable components of Google Cloud:\n",
        "\n",
        "* [Vertex AI](https://cloud.google.com/vertex-ai/pricing)\n",
        "* [Cloud Storage](https://cloud.google.com/storage/pricing)\n",
        "* [Dataproc](https://cloud.google.com/dataproc/pricing)\n",
        "\n",
        "You can use the [Pricing Calculator](https://cloud.google.com/products/calculator/) to generate a cost estimate based on your projected usage."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "4b331e2fd155"
      },
      "source": [
        "## Get Started"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "WReHDGG5g0XY"
      },
      "source": [
        "### Install Vertex AI SDK for Python and other required packages"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "lWEdiXsJg0XY"
      },
      "source": [
        "Since the test environment does not have Java and PySpark, the following cell is necessary for the testing purpose."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "5f0d58c0e522"
      },
      "outputs": [],
      "source": [
        "import os\n",
        "\n",
        "if os.getenv(\"IS_TESTING\"):\n",
        "    \"\"\"\n",
        "    The testing suite does not currently support testing on Dataproc clusters,\n",
        "    so the testing environment is setup to replicate Dataproc via the following steps.\n",
        "    \"\"\"\n",
        "    JAVA_VER = \"8u332-b09\"\n",
        "    JAVA_FOLDER = \"/tmp/java\"\n",
        "    FILE_NAME = f\"openlogic-openjdk-{JAVA_VER}-linux-x64\"\n",
        "    TAR_FILE = f\"{JAVA_FOLDER}/{FILE_NAME}.tar.gz\"\n",
        "    DOWNLOAD_LINK = f\"https://builds.openlogic.com/downloadJDK/openlogic-openjdk/{JAVA_VER}/openlogic-openjdk-{JAVA_VER}-linux-x64.tar.gz\"\n",
        "    PYSPARK_VER = \"3.1.3\"\n",
        "\n",
        "    # Download Open JDK 8. Spark requires Java to execute.\n",
        "    ! rm -rf $JAVA_FOLDER\n",
        "    ! mkdir $JAVA_FOLDER\n",
        "    ! wget -P $JAVA_FOLDER $DOWNLOAD_LINK\n",
        "    os.environ[\"JAVA_HOME\"] = f\"{JAVA_FOLDER}/{FILE_NAME}\"\n",
        "    ! tar -zxf $TAR_FILE -C $JAVA_FOLDER\n",
        "    ! echo $JAVA_HOME\n",
        "\n",
        "    # Pin the Spark version to match that the Dataproc 2.0 cluster.\n",
        "    ! pip install pyspark==$PYSPARK_VER -q"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "16220914acc5"
      },
      "source": [
        "### Restart runtime (Colab only)\n",
        "\n",
        "To use the newly installed packages, you must restart the runtime on Google Colab."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "157953ab28f0"
      },
      "outputs": [],
      "source": [
        "import sys\n",
        "\n",
        "if \"google.colab\" in sys.modules:\n",
        "\n",
        "    import IPython\n",
        "\n",
        "    app = IPython.Application.instance()\n",
        "    app.kernel.do_shutdown(True)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "b96b39fd4d7b"
      },
      "source": [
        "<div class=\"alert alert-block alert-warning\">\n",
        "<b>⚠️ The kernel is going to restart. Wait until it's finished before continuing to the next step. ⚠️</b>\n",
        "</div>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ff666ce4051c"
      },
      "source": [
        "### Authenticate your notebook environment (Colab only)\n",
        "\n",
        "Authenticate your environment on Google Colab."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "cc7251520a07"
      },
      "outputs": [],
      "source": [
        "import sys\n",
        "\n",
        "if \"google.colab\" in sys.modules:\n",
        "\n",
        "    from google.colab import auth\n",
        "\n",
        "    auth.authenticate_user()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "lWEdiXsJg0XY"
      },
      "source": [
        "### Create a Dataproc cluster\n",
        "\n",
        "The Spark job executed in this notebook tutorial is compute intensive. Since the job can take a significant amount time to complete in a standard notebook environment, this notebook tutorial runs on a Dataproc cluster that is created with the Dataproc Component Gateway and Jupyter component installed on the cluster.\n",
        "\n",
        "**Existing Dataproc with Jupyter cluster?**: If you have a running Dataproc cluster that has the [Component Gateway and Jupyter component installed on the cluster](https://cloud.google.com/dataproc/docs/concepts/components/jupyter#gcloud-command), you can use it in this tutorial. If you plan to use it, skip this step, and go to `Switch your kernel`."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "oM1iC_MfAts1"
      },
      "outputs": [],
      "source": [
        "import os\n",
        "\n",
        "if not os.getenv(\"IS_TESTING\"):\n",
        "    CLUSTER_NAME = \"[your-cluster]\"  # @param {type: \"string\"}\n",
        "    CLUSTER_LOCATION = \"[your-location]\"  # @param {type: \"string\"}\n",
        "\n",
        "    if CLUSTER_LOCATION == \"[your-location]\":\n",
        "        CLUSTER_LOCATION = \"us-central1\"\n",
        "\n",
        "    print(f\"CLUSTER_NAME: {CLUSTER_NAME}\")\n",
        "    print(f\"CLUSTER_LOCATION: {CLUSTER_LOCATION}\")"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "oM1iC_MfAts1"
      },
      "outputs": [],
      "source": [
        "# Enable the Dataproc api\n",
        "! gcloud services enable dataproc.googleapis.com\n",
        "\n",
        "if not os.getenv(\"IS_TESTING\"):\n",
        "    !gcloud dataproc clusters create $CLUSTER_NAME \\\n",
        "        --region=$CLUSTER_LOCATION \\\n",
        "        --enable-component-gateway \\\n",
        "        --image-version=2.0 \\\n",
        "        --optional-components=JUPYTER"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "lWEdiXsJg0XY"
      },
      "source": [
        "Your `CLUSTER_NAME` must be **unique within your Google Cloud project**. It must start with a lowercase letter, followed by up to 51 lowercase letters, numbers, and hyphens, and cannot end with a hyphen."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "lWEdiXsJg0XY"
      },
      "source": [
        "#### Switch your kernel\n",
        "\n",
        "Your notebook kernel is listed at the top of the notebook page. Your notebook should run on the Python 3 kernel running on your Dataproc cluster.\n",
        "\n",
        "Select **Kernel > Change Kernel** from the top menu, then select `Python 3 on CLUSTER_NAME: Dataproc cluster in REGION (Remote)`."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "WReHDGG5g0XY"
      },
      "source": [
        "### Set Google Cloud project information\n",
        "\n",
        "To get started using Vertex AI, you must have an existing Google Cloud project. Learn more about [setting up a project and a development environment](https://cloud.google.com/vertex-ai/docs/start/cloud-environment)."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "set_project_id"
      },
      "outputs": [],
      "source": [
        "PROJECT_ID = \"[your-project-id]\"  # @param {type:\"string\"}"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "zgPO1eR3CYjk"
      },
      "source": [
        "### Create a Bigquery Dataset\n",
        "\n",
        "The Spark DataFrame created in this tutorial is stored in BigQuery."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "zgPO1eR3CYjk"
      },
      "source": [
        "Set the name of your BigQuery dataset, then create it."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "MzGDU7TWdts_"
      },
      "outputs": [],
      "source": [
        "if not os.getenv(\"IS_TESTING\"):\n",
        "    DATASET_NAME = \"[your-dataset-name]\"  # @param {type:\"string\"}\n",
        "\n",
        "    if (\n",
        "        DATASET_NAME == \"\"\n",
        "        or DATASET_NAME is None\n",
        "        or DATASET_NAME == \"[your-dataset-name]\"\n",
        "    ):\n",
        "        DATASET_NAME = PROJECT_ID.replace(\"-\", \"_\")\n",
        "        DATASET_NAME = f\"{DATASET_NAME}_unique\"\n",
        "else:\n",
        "    DATASET_NAME = \"python_docs_samples_tests_spark_unique\""
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "vhOb7YnwClBb"
      },
      "outputs": [],
      "source": [
        "! bq mk $DATASET_NAME"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "XoEqT2Y4DJmf"
      },
      "source": [
        "### Import required libraries"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "pRUOFELefqf1"
      },
      "outputs": [],
      "source": [
        "# You use Spark SQL in a \"SparkSession\" to create DataFrames\n",
        "from pyspark.sql import SparkSession\n",
        "# PySpark functions\n",
        "from pyspark.sql.functions import avg, col, count, desc, round, size, udf\n",
        "# These allow us to create a schema for our data\n",
        "from pyspark.sql.types import ArrayType, IntegerType, StringType"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "init_aip:mbsdk,all"
      },
      "source": [
        "### Initialize the SparkSession\n",
        "\n",
        "To use Apache Spark with BigQuery, you must include the [spark-bigquery-connector](https://github.com/GoogleCloudDataproc/spark-bigquery-connector) when you initialize the `SparkSession`."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "init_aip:mbsdk,all"
      },
      "outputs": [],
      "source": [
        "# Initialize the \"SparkSession\" with the following config.\n",
        "VER = \"0.26.0\"\n",
        "FILE_NAME = f\"spark-bigquery-with-dependencies_2.12-{VER}.jar\"\n",
        "\n",
        "if os.getenv(\"IS_TESTING\"):\n",
        "    connector = f\"https://github.com/GoogleCloudDataproc/spark-bigquery-connector/releases/download/{VER}/{FILE_NAME}\"\n",
        "else:\n",
        "    connector = f\"gs://spark-lib/bigquery/{FILE_NAME}\"\n",
        "\n",
        "spark = (\n",
        "    SparkSession.builder.appName(\"spark-bigquery-polyglot-language-demo\")\n",
        "    .config(\"spark.jars\", connector)\n",
        "    .config(\"spark.sql.debug.maxToStringFields\", \"500\")\n",
        "    .getOrCreate()\n",
        ")"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "init_aip:mbsdk,all"
      },
      "source": [
        "### Fetch data from BigQuery"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "init_aip:mbsdk,all"
      },
      "outputs": [],
      "source": [
        "# Load the Github Activity public dataset from BigQuery.\n",
        "df = (\n",
        "    spark.read.format(\"bigquery\")\n",
        "    .option(\"table\", \"bigquery-public-data.github_repos.languages\")\n",
        "    .load()\n",
        ")\n",
        "\n",
        "# Restrict testing data since the testing environment runs on a small Docker image.\n",
        "if os.getenv(\"IS_TESTING\"):\n",
        "    df = df.sample(0.0001)\n",
        "\n",
        "df.printSchema()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "403176b059ae"
      },
      "source": [
        "### Preprocessing\n",
        "\n",
        "As shown in the displayed schema, Github Activity data is stored in arrays, not in primitive types. \n",
        "\n",
        "To work effectively with data, convert the arrays to primitive types, and separate the monoglot and polyglot repo data.\n",
        "\n",
        "The return type of three Python functions have an `@udf` annotation (signifying a [User Defined Function](https://spark.apache.org/docs/3.1.3/api/python/reference/api/pyspark.sql.functions.udf.html)). UDFs extend PySpark framework functions."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "2e56d855da76"
      },
      "outputs": [],
      "source": [
        "# Set the LIMIT constant as 10 to get the top ten results.\n",
        "LIMIT = 10\n",
        "\n",
        "# A constant used to explode the pie chart to aid visibility.\n",
        "EXPLODE_PIE_CHART = tuple([0.05] * LIMIT)\n",
        "\n",
        "\n",
        "@udf(returnType=StringType())\n",
        "def language_to_mono_language(language) -> str:\n",
        "    \"\"\"\n",
        "    The preprocessing function takes a language array and returns its name if the language has one element.\n",
        "    Args:\n",
        "        language: list of struct that contains name and bytes.\n",
        "                  (e.g., language = [[name: \"C\", bytes: 300]]\n",
        "    Returns:\n",
        "        Monorepo's name\n",
        "    \"\"\"\n",
        "    return language[0].name if len(language) == 1 else None\n",
        "\n",
        "\n",
        "@udf(returnType=IntegerType())\n",
        "def language_to_mono_size(language) -> int:\n",
        "    \"\"\"\n",
        "    The preprocessing function takes a language array and returns its bytes if the language has one element.\n",
        "    Args:\n",
        "        language: list of struct that contains name and bytes.\n",
        "                  (e.g., language = [[name: \"C\", bytes: 300]]\n",
        "    Returns:\n",
        "        Monorepo's bytes\n",
        "    \"\"\"\n",
        "    return language[0].bytes if len(language) == 1 else 0\n",
        "\n",
        "\n",
        "@udf(returnType=StringType())\n",
        "def language_to_poly_language(language) -> str:\n",
        "    \"\"\"\n",
        "    The preprocessing function takes a language array and returns the top three language names based on their bytes.\n",
        "    Args:\n",
        "        language: list of struct that contains name and bytes.\n",
        "                  (e.g., language = [[name: \"C\", bytes: 300],\n",
        "                                     [name: \"Java\", bytes: 200]]\n",
        "    Returns:\n",
        "        Polyrepo's name in string form separated by commas\n",
        "    \"\"\"\n",
        "    if len(language) < 2:\n",
        "        return None\n",
        "    # Sort languages by their bytes in a descending order.\n",
        "    language.sort(key=lambda x: -x.bytes)\n",
        "    top_3 = language[:3]\n",
        "\n",
        "    # Sort top_3 languages by their name.\n",
        "    top_3.sort(key=lambda x: x.name)\n",
        "    ret = []\n",
        "    for elem in top_3:\n",
        "        ret.append(elem.name)\n",
        "    return \", \".join(ret)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "2e56d855da76"
      },
      "outputs": [],
      "source": [
        "# Create a DataFrame named \"preprocessed_df\", with the array split into three columns using UDF.\n",
        "preprocessed_df = df.select(\n",
        "    col(\"repo_name\"),\n",
        "    language_to_mono_language(col(\"language\")).alias(\"mono_language\"),\n",
        "    language_to_mono_size(col(\"language\")).alias(\"mono_size\"),\n",
        "    language_to_poly_language(col(\"language\")).alias(\"poly_language\"),\n",
        ")\n",
        "preprocessed_df.printSchema()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "668a2549b231"
      },
      "source": [
        "The displayed `preprocessed_df`'s schema shows the language column separated into three columns: `mono_language`, `mono_size`, and `poly_language`."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "efec0bfbfb65"
      },
      "outputs": [],
      "source": [
        "# Output the number of repositories of monoglot(single language used) and polyglot(multiple languages used).\n",
        "mono = preprocessed_df.where(col(\"mono_language\").isNotNull()).count()\n",
        "print(f\"The number of repositories that use one language is {mono}\")\n",
        "\n",
        "poly = preprocessed_df.where(col(\"poly_language\").isNotNull()).count()\n",
        "print(f\"The number of repositories that use multiple languages is {poly}\")\n",
        "\n",
        "poly_percent = (poly / (mono + poly)) * 100\n",
        "print(\n",
        "    f\"Polyglot repositories comprise approximately {poly_percent:.2f}% of the total number of repositories.\"\n",
        ")"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "0d79c26911f0"
      },
      "source": [
        "### Analyze\n",
        "\n",
        "#### What language is most frequently used in the monoglot repos?\n",
        "To answer this question, execute the following query with the preprocessed column, `mono_language`."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "a241cfdd169d"
      },
      "outputs": [],
      "source": [
        "# Get the monoglot repositories and sort them based on language popularity.\n",
        "mono_ranking = (\n",
        "    preprocessed_df.groupBy(\"mono_language\")\n",
        "    .count()\n",
        "    .sort(desc(\"count\"))\n",
        "    .where(col(\"mono_language\").isNotNull())\n",
        ")\n",
        "mono_ranking.show()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "9e5349dca8e6"
      },
      "source": [
        "Using `mono_ranking`, visualize the results with a pie chart."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "7b9eb5a4f47e"
      },
      "outputs": [],
      "source": [
        "# Convert the Spark DataFrame to a Pandas DataFrame to display the pie chart.\n",
        "mono_panda = mono_ranking.toPandas()[:LIMIT].copy()\n",
        "mono_panda.groupby([\"mono_language\"]).sum().plot(\n",
        "    kind=\"pie\",\n",
        "    y=\"count\",\n",
        "    autopct=\"%1.1f%%\",\n",
        "    label=\"\",\n",
        "    title=\"Monoglot repositories\",\n",
        "    legend=False,\n",
        "    figsize=(7, 7),\n",
        "    explode=EXPLODE_PIE_CHART,\n",
        ")"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "cc68a7c604b4"
      },
      "source": [
        "#### What is the average size of each language in the monoglot repos?\n",
        "\n",
        "Preprocessed the `mono_size` and `mono_language` columns to get the average size of each language.\n",
        "\n",
        "`mono_size` bytes are stated in kilobytes. The following query divides `mono_size` by 1000 to convert the size to megabytes."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "d5d3e2e38df7"
      },
      "outputs": [],
      "source": [
        "mono_ranking_avg_bytes = (\n",
        "    preprocessed_df.groupBy(\"mono_language\")\n",
        "    .agg(\n",
        "        count(\"mono_language\").alias(\"count\"),\n",
        "        round(avg(\"mono_size\") / 1000).alias(\"average_in_MB\"),\n",
        "    )\n",
        "    .sort(desc(\"average_in_MB\"))\n",
        "    .where(col(\"mono_language\").isNotNull() & (col(\"count\") > 500))\n",
        ")\n",
        "\n",
        "mono_ranking_avg_bytes.show()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "4551e1b7b28b"
      },
      "source": [
        "#### What three languages are most frequently found together in polyglot repos?\n",
        "\n",
        "Using the preprocessed `poly_language` column, implement a query to show the ranking of polyglot repositories with the top 3 languages based on size."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "c8285a340a4a"
      },
      "outputs": [],
      "source": [
        "# Get the polyglot repositories by language popularity.\n",
        "poly_ranking = (\n",
        "    preprocessed_df.groupBy(\"poly_language\")\n",
        "    .count()\n",
        "    .sort(desc(\"count\"))\n",
        "    .where(col(\"poly_language\").isNotNull())\n",
        ")\n",
        "\n",
        "poly_ranking.show()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "633a0752b053"
      },
      "source": [
        "The majority of results contain a combination of HTML or CSS, and Javascript.\n",
        "\n",
        "Display a pie chart:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "30ae2d750ab7"
      },
      "outputs": [],
      "source": [
        "# Convert the Spark DataFrame to a Pandas DataFrame to display the pie chart.\n",
        "poly_panda = poly_ranking.toPandas()[:LIMIT].copy()\n",
        "poly_panda.groupby([\"poly_language\"]).sum().plot(\n",
        "    kind=\"pie\",\n",
        "    y=\"count\",\n",
        "    autopct=\"%1.1f%%\",\n",
        "    label=\"\",\n",
        "    title=\"Polyglot repositories\",\n",
        "    legend=False,\n",
        "    figsize=(7, 7),\n",
        "    explode=EXPLODE_PIE_CHART,\n",
        ")"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "3905037bcea3"
      },
      "source": [
        "The pie chart shows that eight out of the top ten results contain either `HTML` or `CSS`.\n",
        "\n",
        "You can create combinations of languages in each repo using the original data fetched from BigQuery."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "25479dcebd93"
      },
      "outputs": [],
      "source": [
        "# A Python package to get combinations.\n",
        "from itertools import combinations\n",
        "# A Python package to use type hint\n",
        "from typing import List\n",
        "\n",
        "# PySpark functions\n",
        "from pyspark.sql.functions import explode"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "ad6a3e548fc3"
      },
      "outputs": [],
      "source": [
        "def normalize_name(name: str) -> str:\n",
        "    \"\"\"\n",
        "    Change the language name to avoid invalid characters in the BigQuery data.\n",
        "    Args:\n",
        "        name: string\n",
        "    Returns:\n",
        "        Normalized name: string\n",
        "    \"\"\"\n",
        "    normalized_arr = []\n",
        "\n",
        "    # The following sets of characters cannot be used in BigQuery's fields.\n",
        "    invalid_chars = {\",\", \";\", \"{\", \"}\", \"(\", \")\", \"\\n\", \"\\t\", \"=\", \"'\"}\n",
        "    replace_chars = {\n",
        "        \" \": \"_\",\n",
        "        \".\": \"_\",\n",
        "        \"-\": \"_\",\n",
        "        \"#\": \"_sharp\",\n",
        "        \"+\": \"_plus\",\n",
        "        \"*\": \"_star\",\n",
        "    }\n",
        "\n",
        "    # The name must start with a letter or underscore.\n",
        "    if name[0].isnumeric():\n",
        "        normalized_arr.append(\"_\")\n",
        "\n",
        "    for ch in name:\n",
        "        # Skip if a character is in the set of invalid characters.\n",
        "        if ch in invalid_chars:\n",
        "            continue\n",
        "\n",
        "        # Replace if a character is in the dictionary of replace_chars.\n",
        "        if ch in replace_chars:\n",
        "            normalized_arr.append(replace_chars[ch])\n",
        "\n",
        "        # Change to lowercase to merge name duplications, for example, \"Java\" and \"java\".\n",
        "        else:\n",
        "            normalized_arr.append(ch.lower())\n",
        "\n",
        "    # Convert the array to string\n",
        "    return \"\".join(normalized_arr)\n",
        "\n",
        "\n",
        "@udf(returnType=ArrayType(StringType()))\n",
        "def reduce_language(language) -> List[str]:\n",
        "    \"\"\"\n",
        "    The preprocess function takes the language and reduces it to remove \"bytes\".\n",
        "    Args:\n",
        "        language: list of struct that contains name and bytes.\n",
        "                  (e.g., language = [[name: \"C\", bytes: 300],\n",
        "                                     [name: \"Java\", bytes: 200]]\n",
        "    Returns:\n",
        "        list of strings that contains name.\n",
        "                  (e.g., reduced_languages = [\"C\", \"Java\"])\n",
        "    \"\"\"\n",
        "    if len(language) < 2:\n",
        "        return None\n",
        "    reduced_languages = []\n",
        "    for elem in language:\n",
        "        # To write back to BigQuery, the name must be normalized.\n",
        "        normalized_name = normalize_name(elem.name)\n",
        "        reduced_languages.append(normalized_name)\n",
        "    return reduced_languages\n",
        "\n",
        "\n",
        "@udf(returnType=ArrayType(ArrayType(StringType())))\n",
        "def preprocess_combination(language) -> List[List[str]]:\n",
        "    \"\"\"\n",
        "    The preprocess function takes the language and returns every language combination.\n",
        "    Args:\n",
        "        language: list of struct that contains name and bytes.\n",
        "                  (e.g., language = [[name: \"C\", bytes: 300],\n",
        "                                     [name: \"Java\", bytes: 200]]\n",
        "    Returns:\n",
        "        List of every possible combinations.\n",
        "                  (e.g., arr_combinations = [[\"C\", \"Java\"], [\"Java\", \"C\"]])\n",
        "    \"\"\"\n",
        "    if not language:\n",
        "        return None\n",
        "    arr_combinations = []\n",
        "    for combination in combinations(language, 2):\n",
        "        arr_combinations.append(combination)\n",
        "        arr_combinations.append(combination[::-1])\n",
        "    return arr_combinations\n",
        "\n",
        "\n",
        "# Preprocess the \"reduced_languages\" column using UDF.\n",
        "df = df.withColumn(\"reduced_languages\", reduce_language(col(\"language\")))\n",
        "\n",
        "# Preprocess the \"combinations\" column using UDF.\n",
        "df = df.withColumn(\"combinations\", preprocess_combination(col(\"reduced_languages\")))\n",
        "\n",
        "# Create another DataFrame from \"df\" that has \"repo_name\" and \"combinations\" as columns.\n",
        "frequency_df = df.select(col(\"repo_name\"), col(\"combinations\")).where(\n",
        "    size(col(\"language\")) > 1\n",
        ")\n",
        "frequency_df.printSchema()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "5e80986d0095"
      },
      "source": [
        "`frequency_df` has the repo name and combinations of languages.\n",
        "\n",
        "Use the Spark `explode()` function, which is similar to the SQL `UNNEST` function.\n",
        "\n",
        "Your table currently has the following content:\n",
        "\n",
        "| repo_name   | combinations      |\n",
        "| :----------: | :---------------: |\n",
        "| a           | [['C', 'C++'], ['C++', 'C'], ['C', 'Java'], ['Java', 'C’], ['C++', 'Java'], ['Java', 'C++']]|\n",
        "| b           | [['C', 'C++'], ['C++', 'C'], ['C', 'Python'], ['Python', 'C'], ['C++', 'Python'], ['Python', 'C++']]|"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "4ec92c4d0652"
      },
      "outputs": [],
      "source": [
        "# explode() converts the elements in combinations to rows.\n",
        "frequency_df = frequency_df.withColumn(\"languages\", explode(col(\"combinations\")))\n",
        "\n",
        "# Create columns for combinations of languages.\n",
        "frequency_df = frequency_df.withColumn(\"language0\", col(\"languages\")[0])\n",
        "frequency_df = frequency_df.withColumn(\"language1\", col(\"languages\")[1])"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "5e80986d0095"
      },
      "source": [
        "After using `explode()` and adding the `langauge0` and `language1` columns, the `frequency_df` table has the following content:\n",
        "\n",
        "| repo_name   | languages         | language0    | language1 |\n",
        "| :---------: | :---------------: | :--------:   | :-------: |\n",
        "| a           | ['C', 'C++']      | 'C'          |'C++'      |\n",
        "| a           | ['C++', 'C']      | 'C++'        |'C'        |\n",
        "| a           | ['C', 'Java']     | 'C'          |'Java'     |\n",
        "| a           | ['Java', 'C’]     | 'Java'       |'C'        |\n",
        "| a           | ['C++', 'Java']   | 'C++'        |'Java'     |\n",
        "| a           | ['Java', 'C++']   | 'Java'       |'C++'      |\n",
        "| b           | ['C', 'C++']      | 'C'          |'C++'      |\n",
        "| b           | ['C++', 'C']      | 'C++'        |'C'        |\n",
        "| b           | ['C', 'Python']   | 'C'          |'Python'   |\n",
        "| b           | ['Python', 'C']   | 'Python'     |'C'        |\n",
        "| b           | ['C++', 'Python'] | 'C++'        |'Python'   |\n",
        "| b           | ['Python', 'C++'] | 'Python'     |'C++'      |\n",
        "\n",
        "Calculate a pair-wise frequency table of the `language0` and `language1` columns. The first column of each row will contain distinct values of `language0`, and column names will contain distinct values of `language1`."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "4ec92c4d0652"
      },
      "outputs": [],
      "source": [
        "# crosstab() reshapes the table into a frequency distribution table by using cross tabulations.\n",
        "frequency_df = frequency_df.crosstab(\"language0\", \"language1\").withColumnRenamed(\n",
        "    \"language0_language1\", \"languages\"\n",
        ")"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "a2c79c4a3326"
      },
      "source": [
        "After applying `crosstab()` to `frequency_df`, the DataFrame data is arranged as follows:\n",
        "\n",
        "| languages  |  C  | C++ | Java | Python |\n",
        "| :--------: | :-: | :-: | :-:  |  :-:   |\n",
        "|     C      |  0  |  2  |  1   |   1    |\n",
        "|     C++    |  2  |  0  |  1   |   1    |\n",
        "|    Java    |  1  |  1  |  0   |   0    |\n",
        "|   Python   |  1  |  1  |  0   |   0    |\n",
        "\n",
        "Note that this table contains sample, not real, data.\n",
        "\n",
        "See [frequency distribution](https://en.wikipedia.org/wiki/Frequency_%28statistics%29#Frequency_distribution_table) and [cross tabulations](https://en.wikipedia.org/wiki/Contingency_table).\n",
        "\n",
        "The DataFrame now contains the frequency of each language. Visualize it with a popular language."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "ee1591ca4563"
      },
      "outputs": [],
      "source": [
        "# Set of popular languages. You can modify this set to show your preferred languages.\n",
        "MAJOR_LANGUAGES = {\"C\", \"Java\", \"Python\", \"JavaScript\", \"Go\"}\n",
        "\n",
        "# Declare a dictionary to store the key as a language name and the value as the selected DataFrame\n",
        "df_dict = dict()\n",
        "\n",
        "for language in MAJOR_LANGUAGES:\n",
        "    # Get a top ten languages of each language and store it to the dictionary.\n",
        "    df_dict[language] = (\n",
        "        frequency_df.select(col(\"languages\"), language).sort(-col(language)).limit(10)\n",
        "    )"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "5951469b9925"
      },
      "outputs": [],
      "source": [
        "for language in df_dict:\n",
        "    # Convert Spark DataFrame to Pandas DataFrame to display the bar chart.\n",
        "    elem_panda = df_dict[language].toPandas()[:LIMIT].copy()\n",
        "    elem_panda.set_index(\"languages\", inplace=True)\n",
        "    elem_panda.sort_values(language, ascending=True, inplace=True)\n",
        "    elem_panda.plot(\n",
        "        kind=\"barh\",\n",
        "        title=language,\n",
        "        legend=False,\n",
        "        xlabel=\"\",\n",
        "    )"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "TpV-iwP9qw9c"
      },
      "source": [
        "### Write back to BigQuery\n",
        "\n",
        "After analyzing these queries, there are several DataFrames: the ranking of monoglot repositories, the average bytes of monoglot repositories, and the frequency table of each language used in a repository. \n",
        "\n",
        "These DataFrames will be stored in BigQuery using the [spark-bigquery-connector](https://github.com/GoogleCloudDataproc/spark-bigquery-connector)."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "b8ab41bf47dd"
      },
      "outputs": [],
      "source": [
        "dataframes = {\n",
        "    \"mono_ranking\": mono_ranking,\n",
        "}\n",
        "if not os.getenv(\"IS_TESTING\"):\n",
        "    dataframes[\"mono_ranking_avg_bytes\"] = mono_ranking_avg_bytes\n",
        "    dataframes[\"frequency_table\"] = frequency_df\n",
        "\n",
        "# Iterate through the DataFrames and save them to the BigQuery.\n",
        "for df in dataframes:\n",
        "    dataframes[df].write.format(\"bigquery\").option(\"writeMethod\", \"direct\").option(\n",
        "        \"table\", f\"{DATASET_NAME}.{df}\"\n",
        "    ).save()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "TpV-iwP9qw9c"
      },
      "source": [
        "If no errors are reported, Congratulations! Your DataFrame is successfully stored in BigQuery.\n",
        "\n",
        "You can view the data in the [Google Cloud console](https://console.corp.google.com/bigquery) or using the `bq` command-line tool."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "e7734cef2fc3"
      },
      "outputs": [],
      "source": [
        "QUERY = f\"SELECT languages, python FROM {PROJECT_ID}.{DATASET_NAME}.frequency_table ORDER BY python DESC LIMIT 10\"\n",
        "\n",
        "! bq query --nouse_legacy_sql $QUERY"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "TpV-iwP9qw9c"
      },
      "source": [
        "## Cleaning up\n",
        "\n",
        "See [Clean up](https://cloud.google.com/vertex-ai/docs/workbench/managed/create-managed-notebooks-instance-console-quickstart#clean-up) to delete your project or the managed notebook created in this tutorial."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "TpV-iwP9qw9c"
      },
      "source": [
        "### Delete BigQuery dataset"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "b77295420ab9"
      },
      "outputs": [],
      "source": [
        "! bq rm -r -f $DATASET_NAME"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "TpV-iwP9qw9c"
      },
      "source": [
        "After you delete the BigQuery dataset, you can check your Datasets in BigQuery using the following command."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "c53e6e169788"
      },
      "outputs": [],
      "source": [
        "! bq ls"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "TpV-iwP9qw9c"
      },
      "source": [
        "### Delete Dataproc Cluster\n",
        "\n",
        "It's not possible to delete the cluster you're currently using unless you switch the kernel to local. To delete it, you need to switch the kernel to local `Python 3` or `PySpark`, set your `CLUSTER_NAME` and `CLUSTER_LOCATION` manually in the following cell, and execute the `gcloud` command.\n",
        "\n",
        "See [Deleting a cluster](https://cloud.google.com/dataproc/docs/guides/manage-cluster#console) to delete the Dataproc cluster created in this tutorial."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "c53e6e169788"
      },
      "outputs": [],
      "source": [
        "CLUSTER_NAME = \"[your-cluster-name]\"\n",
        "CLUSTER_LOCATION = \"[your-cluster-location]\""
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "b8eeeb3cd293"
      },
      "outputs": [],
      "source": [
        "if not os.getenv(\"IS_TESTING\"):\n",
        "    ! gcloud dataproc clusters delete $CLUSTER_NAME --region=$CLUSTER_LOCATION -q"
      ]
    }
  ],
  "metadata": {
    "colab": {
      "collapsed_sections": [],
      "name": "spark_bigquery.ipynb",
      "toc_visible": true
    },
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}