Encoding+dataframe+columns.ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/groda/big_data/blob/master/Encoding%2Bdataframe%2Bcolumns.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "doxTgZtiazqa"
      },
      "source": [
        "<a href=\"https://github.com/groda/big_data\"><div><img src=\"https://github.com/groda/big_data/blob/master/logo_bdb.png?raw=true\" align=right width=\"90\"></div></a>\n",
        "\n",
        "# Encode columns in csv file\n",
        "<br>\n",
        "<br>\n",
        "\n",
        "\n",
        "\n",
        "I'm given a CSV file containing strings and I want to convert the characters to numeric values. I want to use different encodings of the characters on different columns or groups of columns.\n",
        "\n",
        "Let's say for instance that I have two encodings __A__ and __B__:\n",
        " - in encoding __A__ I want to encode the character `a` with the number `1`, the character `b` with `2`, and `c` with `3`\n",
        " - in encoding __B__ I want to encode the character `a` with the number `2`, the character `b` with `3`, and `c` with `1`\n",
        "\n",
        "If I use encoding __A__ to transform all columns in table\n",
        "\n",
        "| c1| c2 |\n",
        "|-----|-----|\n",
        "| a | a|\n",
        "| b | b|\n",
        "| c | b|\n",
        "\n",
        "I obtain\n",
        "\n",
        "| c1_enc| c2_enc |\n",
        "|-----|-----|\n",
        "| 1 | 1|\n",
        "| 2 | 2|\n",
        "| 3 | 2|\n",
        "\n",
        "If `col1` is encoded with __A__ and `col2` is encoded with __A__ then the table becomes\n",
        "\n",
        "| c1_enc| c2_enc |\n",
        "|-----|-----|\n",
        "| 1 | 2|\n",
        "| 2 | 3|\n",
        "| 3 | 3|"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Install PySpark"
      ],
      "metadata": {
        "id": "2z76iZ3bkyLB"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "!pip install pyspark"
      ],
      "metadata": {
        "id": "uzgQcWAOk0Us",
        "outputId": "4c4bcb71-1d7b-4c7c-c871-298503c9e199",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "execution_count": 1,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Collecting pyspark\n",
            "  Downloading pyspark-3.5.1.tar.gz (317.0 MB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m317.0/317.0 MB\u001b[0m \u001b[31m2.8 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25h  Preparing metadata (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "Requirement already satisfied: py4j==0.10.9.7 in /usr/local/lib/python3.10/dist-packages (from pyspark) (0.10.9.7)\n",
            "Building wheels for collected packages: pyspark\n",
            "  Building wheel for pyspark (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "  Created wheel for pyspark: filename=pyspark-3.5.1-py2.py3-none-any.whl size=317488493 sha256=84946858618e84665e32f1a89fc263413fde8ff71e7cc914c960869f264bf835\n",
            "  Stored in directory: /root/.cache/pip/wheels/80/1d/60/2c256ed38dddce2fdd93be545214a63e02fbd8d74fb0b7f3a6\n",
            "Successfully built pyspark\n",
            "Installing collected packages: pyspark\n",
            "Successfully installed pyspark-3.5.1\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Download the data\n",
        "\n",
        "Retrieve the CSV file `data-1600cols.csv` and write it to the local storage."
      ],
      "metadata": {
        "id": "IcTvspsLnB_j"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "import requests\n",
        "import csv\n",
        "\n",
        "def download_csv(url, save_path):\n",
        "    response = requests.get(url)\n",
        "\n",
        "    if response.status_code == 200:\n",
        "        with open(save_path, 'wb') as file:\n",
        "            file.write(response.content)\n",
        "        print(f\"CSV file downloaded successfully and saved at: {save_path}\")\n",
        "    else:\n",
        "        print(f\"Failed to download CSV file. Status code: {response.status_code}\")\n",
        "\n",
        "url = \"https://raw.githubusercontent.com/groda/big_data/master/data-1600cols.csv\"\n",
        "save_path = \"data-1600cols.csv\"\n",
        "\n",
        "download_csv(url, save_path)"
      ],
      "metadata": {
        "id": "J7sbQJsonF6x",
        "outputId": "2b26d8d6-2fba-4779-f4f5-84cb008e1f6a",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "execution_count": 2,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "CSV file downloaded successfully and saved at: data-1600cols.csv\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "cMFSoMcgazqk"
      },
      "source": [
        "## Initialize Spark session\n",
        "\n",
        "SparkContext allows me to access Dataframes, change Spark configuration, cancel a job, get status of a job, etc.\n",
        "\n",
        "Load  the CSV file `data-1600cols.csv` into a Spark dataframe using the file's header as column names."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 3,
      "metadata": {
        "id": "XszooNxRazqs",
        "outputId": "0a9ed61c-990b-4beb-83c0-705d63f89ff5",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stderr",
          "text": [
            "/usr/local/lib/python3.10/dist-packages/pyspark/sql/context.py:113: FutureWarning: Deprecated in 3.0.0. Use SparkSession.builder.getOrCreate() instead.\n",
            "  warnings.warn(\n"
          ]
        }
      ],
      "source": [
        "from pyspark.sql import SparkSession\n",
        "from pyspark import SparkContext\n",
        "from pyspark.sql import SQLContext\n",
        "\n",
        "spark = SparkSession \\\n",
        "            .builder \\\n",
        "            .master(\"local\") \\\n",
        "            .appName(\"Encode multiple columns\") \\\n",
        "            .getOrCreate()\n",
        "\n",
        "sqlContext = SQLContext(spark)\n",
        "df = sqlContext.read.csv(\"data-1600cols.csv\", header=True)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "jzhNAX3Uazqv"
      },
      "source": [
        "Check configuration"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 4,
      "metadata": {
        "id": "E8xrZgjwazqv",
        "outputId": "50f11103-74fd-4131-b27a-1eac525e0590",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "[('spark.master', 'local'),\n",
              " ('spark.driver.extraJavaOptions',\n",
              "  '-Djava.net.preferIPv6Addresses=false -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/jdk.internal.ref=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 -Djdk.reflect.useDirectMethodHandle=false'),\n",
              " ('spark.app.startTime', '1708940663384'),\n",
              " ('spark.driver.host', 'c5f0549691a9'),\n",
              " ('spark.app.submitTime', '1708940662888'),\n",
              " ('spark.executor.id', 'driver'),\n",
              " ('spark.app.id', 'local-1708940668333'),\n",
              " ('spark.sql.warehouse.dir', 'file:/content/spark-warehouse'),\n",
              " ('spark.app.name', 'Encode multiple columns'),\n",
              " ('spark.rdd.compress', 'True'),\n",
              " ('spark.executor.extraJavaOptions',\n",
              "  '-Djava.net.preferIPv6Addresses=false -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/jdk.internal.ref=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 -Djdk.reflect.useDirectMethodHandle=false'),\n",
              " ('spark.serializer.objectStreamReset', '100'),\n",
              " ('spark.submit.pyFiles', ''),\n",
              " ('spark.submit.deployMode', 'client'),\n",
              " ('spark.ui.showConsoleProgress', 'true'),\n",
              " ('spark.driver.port', '45761')]"
            ]
          },
          "metadata": {},
          "execution_count": 4
        }
      ],
      "source": [
        "spark.sparkContext.getConf().getAll()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "rS2J5wfGazqw"
      },
      "source": [
        "Check size of the dataframe (number of rows and columns)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 5,
      "metadata": {
        "id": "0jeYKzP9azqw",
        "outputId": "2f4b5277-750e-4e57-8b54-fb893a704447",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Number of rows: 1000\n",
            "Number of columns: 1600\n"
          ]
        }
      ],
      "source": [
        "print('Number of rows: {}\\nNumber of columns: {}'.format(df.count(),len(df.columns)))"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "a6Rq5kiMazqx"
      },
      "source": [
        "Check if the dataframe contains any nulls?"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 6,
      "metadata": {
        "id": "f-aJhYe1azqx",
        "outputId": "a7a0de89-7eb4-4eeb-e78b-00e2e44d0375",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "[]"
            ]
          },
          "metadata": {},
          "execution_count": 6
        }
      ],
      "source": [
        "df.where(df.V2.isNull()).collect()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "hE0dPL5qazqx"
      },
      "source": [
        "Show a couple of columns"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 7,
      "metadata": {
        "id": "APnPdgpAazqy",
        "outputId": "d6ea1569-151b-4d84-9fbb-c5eefef00af7",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+---+---+---+\n",
            "| V1| V2| V3|\n",
            "+---+---+---+\n",
            "|  j|  n|  d|\n",
            "|  d|  n|  w|\n",
            "|  p|  h|  a|\n",
            "|  b|  h|  e|\n",
            "|  z|  x|  u|\n",
            "|  b|  e|  v|\n",
            "|  y|  t|  x|\n",
            "|  i|  r|  e|\n",
            "|  x|  e|  g|\n",
            "|  l|  j|  z|\n",
            "|  l|  v|  l|\n",
            "|  z|  n|  h|\n",
            "|  s|  m|  c|\n",
            "|  g|  m|  f|\n",
            "|  i|  p|  n|\n",
            "|  i|  f|  b|\n",
            "|  u|  n|  j|\n",
            "|  s|  o|  e|\n",
            "|  k|  y|  c|\n",
            "|  h|  b|  i|\n",
            "+---+---+---+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ],
      "source": [
        "df.select('V1','V2','V3').show()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "FauIiJe1azqy"
      },
      "source": [
        "## First approach\n",
        "\n",
        "Using the `translate` function from `pyspark.sql` and adding a new column with `withColumn` at each step. Test on a small dataframe `test_df`."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 8,
      "metadata": {
        "id": "4DOTSzTKazqy",
        "outputId": "80faee62-82e9-435d-a076-649b2fe29073",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+---+---+\n",
            "| c1| c2|\n",
            "+---+---+\n",
            "|  a|  a|\n",
            "|  b|  b|\n",
            "|  c|  b|\n",
            "+---+---+\n",
            "\n",
            "+---+---+------+------+\n",
            "| c1| c2|c1_enc|c2_enc|\n",
            "+---+---+------+------+\n",
            "|  a|  a|     1|     1|\n",
            "|  b|  b|     2|     2|\n",
            "|  c|  b|     3|     2|\n",
            "+---+---+------+------+\n",
            "\n"
          ]
        }
      ],
      "source": [
        "import pyspark.sql.functions as f\n",
        "\n",
        "test_df = sqlContext.createDataFrame([('a', 'a'), ('b', 'b'), ('c', 'b')], ['c1', 'c2'])\n",
        "test_df.show()\n",
        "\n",
        "chars = \"abc\"\n",
        "A = \"123\" # encoding A\n",
        "B = \"231\" # encoding B\n",
        "\n",
        "\n",
        "for col_name in [\"c1\", \"c2\"]:\n",
        "    test_df = test_df.withColumn(col_name+'_enc', f.translate(f.col(col_name), \"abcd\", A))\n",
        "\n",
        "test_df.show()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "vNf0v2GPazqy"
      },
      "source": [
        "Try out this approach on the big dataframe, applying the function to a few columns. I define two random encodings, `encodingA` and `encodingB` and apply each encoding to two different columns."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 9,
      "metadata": {
        "id": "KLEtwWtfazq0",
        "outputId": "09bd8db1-777c-4bc9-b966-7f742b426f60",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Encodings:\n",
            "abcdefghijklmnopqrstuvwxyz\n",
            "84909340662170830129865816\n",
            "03946914819742444812351068\n",
            "--------------------------\n",
            "+---+---+---+---+------+------+------+------+\n",
            "| V1| V2| V3| V4|V1_enc|V2_enc|V3_enc|V4_enc|\n",
            "+---+---+---+---+------+------+------+------+\n",
            "|  j|  n|  d|  m|     6|     2|     0|     4|\n",
            "|  d|  n|  w|  y|     0|     2|     5|     6|\n",
            "|  p|  h|  a|  h|     3|     4|     8|     4|\n",
            "|  b|  h|  e|  t|     4|     4|     9|     2|\n",
            "|  z|  x|  u|  d|     6|     0|     8|     4|\n",
            "|  b|  e|  v|  j|     4|     6|     6|     1|\n",
            "|  y|  t|  x|  w|     1|     2|     8|     1|\n",
            "|  i|  r|  e|  q|     6|     8|     9|     4|\n",
            "|  x|  e|  g|  s|     8|     6|     4|     1|\n",
            "|  l|  j|  z|  h|     1|     1|     6|     4|\n",
            "|  l|  v|  l|  w|     1|     5|     1|     1|\n",
            "|  z|  n|  h|  z|     6|     2|     0|     8|\n",
            "|  s|  m|  c|  z|     2|     4|     9|     8|\n",
            "|  g|  m|  f|  j|     4|     4|     3|     1|\n",
            "|  i|  p|  n|  h|     6|     4|     0|     4|\n",
            "|  i|  f|  b|  r|     6|     9|     4|     8|\n",
            "|  u|  n|  j|  p|     8|     2|     6|     4|\n",
            "|  s|  o|  e|  f|     2|     4|     9|     9|\n",
            "|  k|  y|  c|  c|     2|     6|     9|     9|\n",
            "|  h|  b|  i|  p|     0|     3|     6|     4|\n",
            "+---+---+---+---+------+------+------+------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ],
      "source": [
        "import string\n",
        "import random\n",
        "\n",
        "# set a raneom seed\n",
        "random.seed(30)\n",
        "\n",
        "chars = string.ascii_lowercase\n",
        "encodingA = ''.join(random.choice(string.digits) for i in range(len(chars)))\n",
        "encodingB = ''.join(random.choice(string.digits) for i in range(len(chars)))\n",
        "\n",
        "print(\"Encodings:\")\n",
        "print(chars)\n",
        "print(encodingA)\n",
        "print(encodingB)\n",
        "print(\"-\"*26)\n",
        "new_df=df\n",
        "\n",
        "for col_name in [\"V1\", \"V3\"]:  # apply encodingA to columns V1, V3\n",
        "    new_df=new_df.withColumn(col_name+'_enc',f.translate(f.col(col_name), chars, encodingA))\n",
        "for col_name in [\"V2\", \"V4\"]:  # apply encodingB to columns V2, V4\n",
        "    new_df=new_df.withColumn(col_name+'_enc',f.translate(f.col(col_name), chars, encodingB))\n",
        "\n",
        "new_df.select(\"V1\",\"V2\",\"V3\",\"V4\", \"V1_enc\", \"V2_enc\", \"V3_enc\", \"V4_enc\").show()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "brjqW1G6azq0"
      },
      "source": [
        "Apply encodings to 4 columns"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 10,
      "metadata": {
        "id": "Jx2C8uhTazq1",
        "outputId": "efd858d0-e9af-4774-e47d-d8b9724f0624",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+---+---+---+---+\n",
            "| V1| V2| V3| V4|\n",
            "+---+---+---+---+\n",
            "|  6|  2|  0|  4|\n",
            "|  0|  2|  5|  6|\n",
            "|  3|  4|  8|  4|\n",
            "+---+---+---+---+\n",
            "only showing top 3 rows\n",
            "\n"
          ]
        }
      ],
      "source": [
        "new_df=df\n",
        "\n",
        "for col_name in [\"V1\", \"V3\"]:  # apply encodingA to columns V1, V2\n",
        "    new_df = new_df.withColumn(col_name,f.translate(f.col(col_name), chars, encodingA))\n",
        "for col_name in [\"V2\", \"V4\"]:  # apply encodingB to columns V3, V4\n",
        "    new_df = new_df.withColumn(col_name,f.translate(f.col(col_name), chars, encodingB))\n",
        "\n",
        "new_df.select(\"V1\",\"V2\",\"V3\",\"V4\").show(3)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "bJoUhtH6azq1"
      },
      "source": [
        "Check:\n",
        "\n",
        "\n",
        "| V1 | V2 | V3 | V4\n",
        "|---|---|---|---|\n",
        "| 6 | 2 | 0 | 4 |\n",
        "| 0 | 2 | 5 | 6 |\n",
        "| 3 | 4 | 8 | 4 |"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "1L6RlfYmazq1"
      },
      "source": [
        "When applying encoding to thousands of rows the previous approach is too slow. The reason is that I'm writing a new dataframe after each tranformation.\n",
        "\n",
        "Split columns in even and odd, apply two different encodings to each set of columns."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 11,
      "metadata": {
        "id": "bZBw3d52azq1",
        "outputId": "5d01e70d-48e3-445c-c664-93eabb0e824a",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "['V2', 'V4']\n",
            "['V1', 'V3']\n",
            "+---+---+---+---+\n",
            "| V1| V2| V3| V4|\n",
            "+---+---+---+---+\n",
            "|  6|  2|  0|  4|\n",
            "|  0|  2|  5|  6|\n",
            "|  3|  4|  8|  4|\n",
            "+---+---+---+---+\n",
            "only showing top 3 rows\n",
            "\n"
          ]
        }
      ],
      "source": [
        "cols_e = [\"V\"+str(i) for i in range(2,5,2)]\n",
        "cols_o = [\"V\"+str(i) for i in range(1,4,2)]\n",
        "\n",
        "print(cols_e)\n",
        "print(cols_o)\n",
        "\n",
        "new_df=df\n",
        "\n",
        "# works with a few columns (4 in total in this example) but too slow for thousands of columns\n",
        "for col_name in cols_o:  # apply encodingA to columns with even numbers\n",
        "    new_df=new_df.withColumn(col_name,f.translate(f.col(col_name), chars, encodingA))\n",
        "for col_name in cols_e:  # apply encodingB to odd columns\n",
        "    new_df=new_df.withColumn(col_name,f.translate(f.col(col_name), chars, encodingB))\n",
        "\n",
        "new_df.select([\"V\"+str(i) for i in range(1,5)]).show(3)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "7ZW9OIB4azq2"
      },
      "source": [
        "## Second approach\n",
        "Using `udf` (user-defined functions). Avoiding `withColumn` and using `select` instead."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 12,
      "metadata": {
        "id": "XSAPE2Whazq2",
        "outputId": "cf87b281-a128-440b-a871-a2d384e34514",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "['V1', 'V3', 'V5']\n",
            "+---+---+---+------+------+------+\n",
            "| V1| V3| V5|V1_enc|V3_enc|V5_enc|\n",
            "+---+---+---+------+------+------+\n",
            "|  j|  d|  s|     6|     0|     2|\n",
            "|  d|  w|  l|     0|     5|     1|\n",
            "|  p|  a|  w|     3|     8|     5|\n",
            "|  b|  e|  x|     4|     9|     8|\n",
            "|  z|  u|  b|     6|     8|     4|\n",
            "|  b|  v|  u|     4|     6|     8|\n",
            "|  y|  x|  z|     1|     8|     6|\n",
            "|  i|  e|  k|     6|     9|     2|\n",
            "|  x|  g|  s|     8|     4|     2|\n",
            "|  l|  z|  l|     1|     6|     1|\n",
            "+---+---+---+------+------+------+\n",
            "only showing top 10 rows\n",
            "\n"
          ]
        }
      ],
      "source": [
        "from pyspark.sql.functions import udf\n",
        "from pyspark.sql.types import IntegerType, StringType\n",
        "\n",
        "# define an encoding as a list of two strings of equal length\n",
        "\n",
        "o = [\"abcdefghijklmnopqrstuvwxyz\", encodingA]\n",
        "\n",
        "def enc(*a):\n",
        "    # encode string s with encoding o\n",
        "    s=a[0]\n",
        "    for i in range(len(o[0])):\n",
        "      if s==o[0][i]:\n",
        "          return o[1][i]\n",
        "    return s\n",
        "\n",
        "# create udf\n",
        "encode_udf = udf(enc, StringType())\n",
        "\n",
        "cols_o = [\"V\"+str(i) for i in range(7) if i%2==1]\n",
        "print(cols_o)\n",
        "\n",
        "(\n",
        "df.select(\"V1\",\"V3\",\"V5\",\n",
        "           encode_udf(\"V1\").alias(\"V1_enc\"),\n",
        "           encode_udf(\"V3\").alias(\"V3_enc\"),\n",
        "           encode_udf(\"V5\").alias(\"V5_enc\"))\n",
        "    .show(10)\n",
        ")\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "pQs1KagOazq3"
      },
      "source": [
        "And now encode all even and odd numbered columns with `encodingA` and `encodingB`, respectively using `select`."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 13,
      "metadata": {
        "id": "lUmXnAdgazq3",
        "outputId": "3c566fbc-4de8-48ba-a60f-dd9acfd5a2f3",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+------+------+------+------+------+-------+-------+-------+-------+-------+\n",
            "|V1_enc|V3_enc|V5_enc|V7_enc|V9_enc|V11_enc|V13_enc|V15_enc|V17_enc|V19_enc|\n",
            "+------+------+------+------+------+-------+-------+-------+-------+-------+\n",
            "|     6|     0|     2|     6|     9|      8|      2|      2|      3|      6|\n",
            "|     0|     5|     1|     8|     0|      2|      9|      6|      8|      2|\n",
            "|     3|     8|     5|     4|     8|      3|      9|      0|      2|      9|\n",
            "|     4|     9|     8|     0|     9|      0|      9|      2|      8|      0|\n",
            "|     6|     8|     4|     0|     9|      2|      8|      6|      6|      6|\n",
            "|     4|     6|     8|     5|     8|      6|      5|      6|      6|      4|\n",
            "|     1|     8|     6|     0|     4|      8|      4|      5|      5|      1|\n",
            "|     6|     9|     2|     5|     8|      8|      5|      4|      0|      1|\n",
            "|     8|     4|     2|     2|     2|      2|      9|      7|      8|      0|\n",
            "|     1|     6|     1|     9|     6|      2|      6|      1|      1|      2|\n",
            "+------+------+------+------+------+-------+-------+-------+-------+-------+\n",
            "only showing top 10 rows\n",
            "\n"
          ]
        }
      ],
      "source": [
        "# apply function to 50 columns\n",
        "new_df=df.select([encode_udf(\"V\"+str(i)).alias(\"V\"+str(i)+\"_enc\") for i in range(1,100,2)])\n",
        "new_df.select([\"V\"+str(i)+\"_enc\" for i in range(1,21,2)]).show(10)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 14,
      "metadata": {
        "id": "yeAi0N8Aazq3",
        "outputId": "8837f69f-b594-46b0-b5c0-79b59f594eb7",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+------+------+------+------+------+-------+-------+-------+-------+-------+\n",
            "|V1_enc|V3_enc|V5_enc|V7_enc|V9_enc|V11_enc|V13_enc|V15_enc|V17_enc|V19_enc|\n",
            "+------+------+------+------+------+-------+-------+-------+-------+-------+\n",
            "|     6|     0|     2|     6|     9|      8|      2|      2|      3|      6|\n",
            "|     0|     5|     1|     8|     0|      2|      9|      6|      8|      2|\n",
            "|     3|     8|     5|     4|     8|      3|      9|      0|      2|      9|\n",
            "|     4|     9|     8|     0|     9|      0|      9|      2|      8|      0|\n",
            "|     6|     8|     4|     0|     9|      2|      8|      6|      6|      6|\n",
            "|     4|     6|     8|     5|     8|      6|      5|      6|      6|      4|\n",
            "|     1|     8|     6|     0|     4|      8|      4|      5|      5|      1|\n",
            "|     6|     9|     2|     5|     8|      8|      5|      4|      0|      1|\n",
            "|     8|     4|     2|     2|     2|      2|      9|      7|      8|      0|\n",
            "|     1|     6|     1|     9|     6|      2|      6|      1|      1|      2|\n",
            "+------+------+------+------+------+-------+-------+-------+-------+-------+\n",
            "only showing top 10 rows\n",
            "\n"
          ]
        }
      ],
      "source": [
        "# apply function to 100 columns\n",
        "new_df=df.select([encode_udf(\"V\"+str(i)).alias(\"V\"+str(i)+\"_enc\") for i in range(1,201,2)])\n",
        "new_df.select([\"V\"+str(i)+\"_enc\" for i in range(1,21,2)]).show(10)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 15,
      "metadata": {
        "id": "Cafg1tG2azq3",
        "outputId": "1e0d9cdb-cbd7-4fe3-e94b-842209f18c08",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+--------+--------+--------+--------+--------+--------+--------+--------+--------+--------+\n",
            "|V381_enc|V383_enc|V385_enc|V387_enc|V389_enc|V391_enc|V393_enc|V395_enc|V397_enc|V399_enc|\n",
            "+--------+--------+--------+--------+--------+--------+--------+--------+--------+--------+\n",
            "|       4|       8|       6|       8|       9|       6|       8|       8|       6|       8|\n",
            "|       6|       4|       8|       4|       1|       6|       6|       0|       9|       6|\n",
            "|       9|       6|       8|       8|       2|       4|       0|       5|       9|       1|\n",
            "|       3|       6|       6|       6|       1|       9|       0|       0|       4|       4|\n",
            "|       6|       0|       1|       0|       1|       1|       2|       9|       2|       8|\n",
            "|       0|       4|       3|       4|       8|       4|       8|       6|       2|       1|\n",
            "|       1|       1|       8|       2|       6|       2|       1|       5|       1|       4|\n",
            "|       4|       4|       1|       0|       1|       2|       4|       8|       8|       8|\n",
            "|       2|       0|       8|       6|       8|       0|       6|       6|       7|       6|\n",
            "|       1|       4|       0|       9|       8|       6|       6|       0|       2|       9|\n",
            "+--------+--------+--------+--------+--------+--------+--------+--------+--------+--------+\n",
            "only showing top 10 rows\n",
            "\n"
          ]
        }
      ],
      "source": [
        "# apply function to 400 columns\n",
        "new_df=df.select([encode_udf(\"V\"+str(i)).alias(\"V\"+str(i)+\"_enc\") for i in range(1,401,2)])\n",
        "new_df.select([\"V\"+str(i)+\"_enc\" for i in range(381,401,2)]).show(10)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 16,
      "metadata": {
        "id": "F-Z8xA6Sazq3",
        "outputId": "ed44c8ff-945b-41ef-e9a1-c29ade410da3",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+--------+--------+--------+--------+--------+--------+--------+--------+--------+--------+\n",
            "|V781_enc|V783_enc|V785_enc|V787_enc|V789_enc|V791_enc|V793_enc|V795_enc|V797_enc|V799_enc|\n",
            "+--------+--------+--------+--------+--------+--------+--------+--------+--------+--------+\n",
            "|       2|       8|       0|       9|       9|       3|       0|       2|       6|       8|\n",
            "|       0|       9|       3|       5|       9|       9|       0|       1|       5|       9|\n",
            "|       6|       4|       8|       8|       3|       8|       5|       0|       3|       0|\n",
            "|       4|       7|       0|       6|       2|       1|       0|       6|       0|       4|\n",
            "|       3|       3|       7|       6|       8|       8|       6|       4|       0|       6|\n",
            "|       8|       8|       1|       8|       8|       4|       4|       5|       4|       2|\n",
            "|       9|       0|       8|       2|       0|       0|       6|       0|       6|       2|\n",
            "|       1|       2|       5|       6|       6|       9|       2|       7|       6|       0|\n",
            "|       0|       8|       2|       0|       8|       1|       7|       9|       8|       1|\n",
            "|       7|       0|       4|       2|       4|       8|       1|       6|       6|       4|\n",
            "+--------+--------+--------+--------+--------+--------+--------+--------+--------+--------+\n",
            "only showing top 10 rows\n",
            "\n"
          ]
        }
      ],
      "source": [
        "# apply function to all odd columns\n",
        "\n",
        "new_df = df.select([encode_udf(\"V\"+str(i)).alias(\"V\"+str(i)+\"_enc\") for i in range(1,801,2)])\n",
        "\n",
        "new_df.select([\"V\"+str(i)+\"_enc\" for i in range(781,801,2)]).show(10)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "imM1VlIGazq4"
      },
      "source": [
        "Now I want to apply different udfs"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 17,
      "metadata": {
        "id": "D8jQgRjOazq4",
        "outputId": "8932e2fd-40af-462a-993c-df91f3754059",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+------+------+------+------+--------+--------+--------+--------+--------+--------+\n",
            "|V1_enc|V2_enc|V3_enc|V4_enc|V795_enc|V796_enc|V797_enc|V798_enc|V799_enc|V800_enc|\n",
            "+------+------+------+------+--------+--------+--------+--------+--------+--------+\n",
            "|     6|     2|     0|     4|       2|       4|       6|       4|       8|       8|\n",
            "|     0|     2|     5|     6|       1|       5|       5|       6|       9|       3|\n",
            "|     3|     4|     8|     4|       0|       0|       3|       2|       0|       6|\n",
            "|     4|     4|     9|     2|       6|       4|       0|       4|       4|       8|\n",
            "|     6|     0|     8|     4|       4|       4|       0|       6|       6|       4|\n",
            "|     4|     6|     6|     1|       5|       5|       4|       9|       2|       1|\n",
            "|     1|     2|     8|     1|       0|       4|       6|       8|       2|       4|\n",
            "|     6|     8|     9|     4|       7|       9|       6|       4|       0|       1|\n",
            "|     8|     6|     4|     1|       9|       8|       8|       8|       1|       3|\n",
            "|     1|     1|     6|     4|       6|       6|       6|       3|       4|       8|\n",
            "+------+------+------+------+--------+--------+--------+--------+--------+--------+\n",
            "only showing top 10 rows\n",
            "\n"
          ]
        }
      ],
      "source": [
        "o = [\"abcdefghijklmnopqrstuvwxyz\", encodingA]\n",
        "e = [\"abcdefghijklmnopqrstuvwxyz\", encodingB]\n",
        "\n",
        "# define two encoding functions\n",
        "\n",
        "def enc1(*a):\n",
        "    # encode string s with encoding o\n",
        "    s=a[0]\n",
        "    for i in range(len(o[0])):\n",
        "      if s==o[0][i]:\n",
        "          return o[1][i]\n",
        "    return s\n",
        "\n",
        "def enc2(*a):\n",
        "    # encode string s with encoding e\n",
        "    s=a[0]\n",
        "    for i in range(len(e[0])):\n",
        "      if s==e[0][i]:\n",
        "          return e[1][i]\n",
        "    return s\n",
        "\n",
        "# create udfs\n",
        "encode_udf1 = udf(enc1, StringType())\n",
        "encode_udf2 = udf(enc2, StringType())\n",
        "\n",
        "new_df = df.select([encode_udf1(\"V\"+str(i)).alias(\"V\"+str(i)+\"_enc\") for i in range(1,800,2)]+\n",
        "                  [encode_udf2(\"V\"+str(i)).alias(\"V\"+str(i)+\"_enc\") for i in range(2,801,2)])\n",
        "new_df.select([\"V\"+str(i)+\"_enc\" for i in range(1,5)]+[\"V\"+str(i)+\"_enc\" for i in range(795,801)]).show(10)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "iG2XLoodazq4"
      },
      "source": [
        "## Export dataframe to file"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 18,
      "metadata": {
        "id": "JFY5crnEazq4",
        "outputId": "92039bc9-0ff5-4896-fee3-4a8b92c6f8e7",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "saved out20240226094527.csv\n"
          ]
        }
      ],
      "source": [
        "import time\n",
        "timestamp = time.strftime(\"%Y%m%d%H%M%S\")\n",
        "new_df.write.csv('out'+timestamp+'.csv', sep=',')\n",
        "print('saved out{}.csv'.format(timestamp))"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "14QZQTn8azq4"
      },
      "source": [
        "Save to CSV with headers"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 19,
      "metadata": {
        "id": "rVhF6djCazq5",
        "outputId": "d0be0d11-1436-4aa8-a41b-af19fad15ae0",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "saved out20240226094539.csv\n"
          ]
        }
      ],
      "source": [
        "timestamp = time.strftime(\"%Y%m%d%H%M%S\")\n",
        "new_df.write.csv('out'+timestamp+'.csv', sep=',', header = True)\n",
        "print('saved out{}.csv'.format(timestamp))"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 20,
      "metadata": {
        "id": "wjpca67Aazq5",
        "outputId": "8b9262ab-74a6-49fd-e206-e594fb4177f1",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "out20240226094527.csv:\n",
            "part-00000-25a5b2d1-15d7-41c0-9d5c-8c5b95a7f198-c000.csv  _SUCCESS\n",
            "\n",
            "out20240226094539.csv:\n",
            "part-00000-ff1b0bcc-8895-4d57-a54c-f03a0c014dce-c000.csv  _SUCCESS\n"
          ]
        }
      ],
      "source": [
        "!ls out*"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "rSZ204Itazq4"
      },
      "source": [
        "## Useful commands for checking system resources\n",
        "\n",
        "The `free -h` and `lscpu` commands are useful for retrieving information about system resources in a Linux environment."
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "The `free -h` command displays information about the system's memory usage in human-readable format. With the `-h` option the command displays sizes in a more human-readable format, using units such as megabytes (MB) and gigabytes (GB) in place of bytes."
      ],
      "metadata": {
        "id": "9CjM3Usor83C"
      }
    },
    {
      "cell_type": "code",
      "execution_count": 21,
      "metadata": {
        "id": "V7v56l91azq4",
        "outputId": "70a1e01f-b2ce-437a-913e-fee539bc3c04",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "               total        used        free      shared  buff/cache   available\n",
            "Mem:            12Gi       1.7Gi       5.2Gi       1.0Mi       5.8Gi        10Gi\n",
            "Swap:             0B          0B          0B\n"
          ]
        }
      ],
      "source": [
        "!free -h"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "The `lscpu` command displays detailed information about the CPU architecture."
      ],
      "metadata": {
        "id": "dxImTX0tsm13"
      }
    },
    {
      "cell_type": "code",
      "execution_count": 22,
      "metadata": {
        "id": "GgbOX557azq4",
        "outputId": "78a8c906-9ca9-41a5-d0f6-ef2a53ca602a",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Architecture:            x86_64\n",
            "  CPU op-mode(s):        32-bit, 64-bit\n",
            "  Address sizes:         46 bits physical, 48 bits virtual\n",
            "  Byte Order:            Little Endian\n",
            "CPU(s):                  2\n",
            "  On-line CPU(s) list:   0,1\n",
            "Vendor ID:               GenuineIntel\n",
            "  Model name:            Intel(R) Xeon(R) CPU @ 2.20GHz\n",
            "    CPU family:          6\n",
            "    Model:               79\n",
            "    Thread(s) per core:  2\n",
            "    Core(s) per socket:  1\n",
            "    Socket(s):           1\n",
            "    Stepping:            0\n",
            "    BogoMIPS:            4399.99\n",
            "    Flags:               fpu vme de pse tsc msr pae mce cx8 apic sep mtrr pge mca cmov pat pse36 clf\n",
            "                         lush mmx fxsr sse sse2 ss ht syscall nx pdpe1gb rdtscp lm constant_tsc rep_\n",
            "                         good nopl xtopology nonstop_tsc cpuid tsc_known_freq pni pclmulqdq ssse3 fm\n",
            "                         a cx16 pcid sse4_1 sse4_2 x2apic movbe popcnt aes xsave avx f16c rdrand hyp\n",
            "                         ervisor lahf_lm abm 3dnowprefetch invpcid_single ssbd ibrs ibpb stibp fsgsb\n",
            "                         ase tsc_adjust bmi1 hle avx2 smep bmi2 erms invpcid rtm rdseed adx smap xsa\n",
            "                         veopt arat md_clear arch_capabilities\n",
            "Virtualization features: \n",
            "  Hypervisor vendor:     KVM\n",
            "  Virtualization type:   full\n",
            "Caches (sum of all):     \n",
            "  L1d:                   32 KiB (1 instance)\n",
            "  L1i:                   32 KiB (1 instance)\n",
            "  L2:                    256 KiB (1 instance)\n",
            "  L3:                    55 MiB (1 instance)\n",
            "NUMA:                    \n",
            "  NUMA node(s):          1\n",
            "  NUMA node0 CPU(s):     0,1\n",
            "Vulnerabilities:         \n",
            "  Gather data sampling:  Not affected\n",
            "  Itlb multihit:         Not affected\n",
            "  L1tf:                  Mitigation; PTE Inversion\n",
            "  Mds:                   Vulnerable; SMT Host state unknown\n",
            "  Meltdown:              Vulnerable\n",
            "  Mmio stale data:       Vulnerable\n",
            "  Retbleed:              Vulnerable\n",
            "  Spec rstack overflow:  Not affected\n",
            "  Spec store bypass:     Vulnerable\n",
            "  Spectre v1:            Vulnerable: __user pointer sanitization and usercopy barriers only; no swap\n",
            "                         gs barriers\n",
            "  Spectre v2:            Vulnerable, IBPB: disabled, STIBP: disabled, PBRSB-eIBRS: Not affected\n",
            "  Srbds:                 Not affected\n",
            "  Tsx async abort:       Vulnerable\n"
          ]
        }
      ],
      "source": [
        "!lscpu"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "In the context of distributed computing, specific values provided by the lscpu command are of particular interest:\n",
        "\n",
        "*   the number of CPUs\n",
        "*   cores per socket\n",
        "*   threads per core\n",
        "*   sockets\n",
        "\n",
        "Understanding these parameters is crucial for assessing the system's potential parallelism.\n",
        "\n",
        "Sockets represents the number of physical processors. Each processor can have one or more cores and each core can execute one or two threads concurrently.\n",
        "\n",
        "Finally, the number of CPUs indicates the total count of independent processing units within each CPU. This is the theoretical upper limit on the number of tasks that can be executed concurrently, offering valuable information for maximizing computational efficiency in distributed computing scenarios.\n",
        "\n",
        "For instance, if you have\n",
        "\n",
        "```\n",
        "Thread(s) per core:    2\n",
        "Core(s) per socket:    4\n",
        "Socket(s):             1\n",
        "```\n",
        "\n",
        "then the total number of independent processing units is\n",
        "\n",
        "$$ 1 × 4 × 2 = 8$$\n",
        "\n",
        "See also: [How many physical CPUs does my machine have?](https://superuser.com/questions/1691479/how-many-physical-cpus-does-my-machine-have).\n",
        "\n"
      ],
      "metadata": {
        "id": "ddxpdot4tSlj"
      }
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    },
    "language_info": {
      "codemirror_mode": {
        "name": "ipython",
        "version": 3
      },
      "file_extension": ".py",
      "mimetype": "text/x-python",
      "name": "python",
      "nbconvert_exporter": "python",
      "pygments_lexer": "ipython3",
      "version": "3.6.2"
    },
    "colab": {
      "provenance": [],
      "toc_visible": true,
      "include_colab_link": true
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}