lab002.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "lab002.ipynb",
      "provenance": [],
      "authorship_tag": "ABX9TyMictcd4hOOq3x4fbaKgXeK",
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/matteolimoncini/Architectures_For_Big_Data/blob/main/lab002.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "L9I3-CxLqAEh"
      },
      "source": [
        ""
      ],
      "execution_count": 14,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "0ggBnn7MqVH8"
      },
      "source": [
        "# 1.MSEG table\n",
        "Find information here: https://www.leanx.eu/en/sap/table/mseg.html.\n",
        "MSEG represent the warehouse status.\n",
        "To let you understand better the idea of CDC, the full extraction from three different days are provided in three different file.\n",
        "\n",
        "\n",
        "\n",
        "First part of the exercise, play with data:"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "%%capture\n",
        "!pip install pyspark"
      ],
      "metadata": {
        "id": "7_ES07hdS95H"
      },
      "execution_count": 15,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "import pyspark\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "from datetime import datetime"
      ],
      "metadata": {
        "id": "dhVljjCRTArW"
      },
      "execution_count": 16,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "sc = pyspark.sql.SparkSession.builder.getOrCreate()"
      ],
      "metadata": {
        "id": "fe1jv6s5TC1i"
      },
      "execution_count": 17,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "## 1. Create a function to read the dataset of a specific day"
      ],
      "metadata": {
        "id": "ZXTfoaz-Tsob"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "def read_day(date):\n",
        "  path = '/content/Lab002_'+date+'.csv'\n",
        "  oneDayRdd = sc.read.csv(path, header=True).rdd\n",
        "  return oneDayRdd"
      ],
      "metadata": {
        "id": "p8ULpEenT0Uf"
      },
      "execution_count": 18,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "## 2. use aforementioned function to read the status of 2021-11-21"
      ],
      "metadata": {
        "id": "B7PqxRsnUWUz"
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "insert date like 'YYYYMMDD'"
      ],
      "metadata": {
        "id": "-4EOGs-FWh8T"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "rdd_21_11_21 = read_day('20211121')"
      ],
      "metadata": {
        "id": "Mtr1gIsYUxWI"
      },
      "execution_count": 19,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "rdd_21_11_21.first()"
      ],
      "metadata": {
        "id": "w-QSvAvMVGRR",
        "outputId": "091016d7-0e1e-4d4c-c87e-93eeffe61843",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "execution_count": 20,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "Row(LGORT='PR08', MATNR='9660906', MEINS='ZT', MENGE='8')"
            ]
          },
          "metadata": {},
          "execution_count": 20
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ioKB3yN9qsPq"
      },
      "source": [
        "## 3. create the Status RDD\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "lCSUcp5xqzDJ"
      },
      "source": [
        "## 4. map each line to have a dictionary\n"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "14gqi46mqvYv"
      },
      "source": [
        "rdd_as_dict = rdd_21_11_21.map(lambda x: pyspark.sql.Row.asDict(x))"
      ],
      "execution_count": 21,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Ib6t0y0Vq2Xi"
      },
      "source": [
        "## 5. show the first line\n"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "cKg8mo6Iq4Fe",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "3a1a4607-62f6-4e06-b55e-427ffbcc6247"
      },
      "source": [
        "rdd_as_dict.first()"
      ],
      "execution_count": 22,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "{'LGORT': 'PR08', 'MATNR': '9660906', 'MEINS': 'ZT', 'MENGE': '8'}"
            ]
          },
          "metadata": {},
          "execution_count": 22
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "yntebN10q6oM"
      },
      "source": [
        "## 6. build a cdc logic in Apache Spark to write back on the disk only fresh lines. \n",
        "Feel free to write on top any additional file to keep the status."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "1o1DgfoMq70Y"
      },
      "source": [
        ""
      ],
      "execution_count": 22,
      "outputs": []
    }
  ]
}