for_hyukjin.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "57cfab78-2493-486a-b67e-d8c041bc8340",
   "metadata": {},
   "source": [
    "## Incremental Append-only Bronze Table\n",
    "#### When \"Phase\" == 20, a Cycle Ends and a New Cycle Starts"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "0ec30909-07bc-4e4c-98fa-84ece7c22245",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+-------+------+\n",
      "|           DateTime|    Key| Value|\n",
      "+-------------------+-------+------+\n",
      "|2022-01-01 10:01:01|  Phase|  20.0|\n",
      "|2022-01-01 10:01:02|SensorA|1103.2|\n",
      "|2022-01-01 10:01:02|SensorB|   1.3|\n",
      "|2022-01-01 10:05:03|  Phase|  80.0|\n",
      "|2022-01-01 10:05:03|SensorA|1107.7|\n",
      "|2022-01-01 10:05:04|SensorB|   1.4|\n",
      "|2022-01-01 10:11:05|  Phase|  20.0|\n",
      "|2022-01-01 10:11:06|SensorA|1108.1|\n",
      "|2022-01-01 10:11:07|SensorB|   1.6|\n",
      "|2022-01-01 10:16:03|  Phase|  80.0|\n",
      "|2022-01-01 10:16:03|SensorA|1109.3|\n",
      "|2022-01-01 10:16:04|SensorB|   1.5|\n",
      "|2022-01-01 10:21:05|  Phase|  20.0|\n",
      "|2022-01-01 10:21:06|SensorA|1110.1|\n",
      "|2022-01-01 10:21:07|SensorB|   1.7|\n",
      "+-------------------+-------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql import functions as F\n",
    "from pyspark.sql import types as T\n",
    "from pyspark.sql.window import Window as W\n",
    "\n",
    "import datetime\n",
    "\n",
    "spark = SparkSession.builder.appName(\"DummyTable\").getOrCreate()\n",
    "\n",
    "schema = T.StructType(\n",
    "    [\n",
    "        T.StructField(\"DateTime\", T.TimestampType(), True),\n",
    "        T.StructField(\"Key\", T.StringType(), True),\n",
    "        T.StructField(\"Value\", T.DoubleType(), True),\n",
    "    ]\n",
    ")\n",
    "\n",
    "input_data = [\n",
    "    (datetime.datetime(2022, 1, 1, 10, 1, 1), \"Phase\", 20.0),\n",
    "    (datetime.datetime(2022, 1, 1, 10, 1, 2), \"SensorA\", 1103.2),\n",
    "    (datetime.datetime(2022, 1, 1, 10, 1, 2), \"SensorB\", 1.3),\n",
    "    (datetime.datetime(2022, 1, 1, 10, 5, 3), \"Phase\", 80.0),\n",
    "    (datetime.datetime(2022, 1, 1, 10, 5, 3), \"SensorA\", 1107.7),\n",
    "    (datetime.datetime(2022, 1, 1, 10, 5, 4), \"SensorB\", 1.4),\n",
    "    (datetime.datetime(2022, 1, 1, 10, 11, 5), \"Phase\", 20.0),\n",
    "    (datetime.datetime(2022, 1, 1, 10, 11, 6), \"SensorA\", 1108.1),\n",
    "    (datetime.datetime(2022, 1, 1, 10, 11, 7), \"SensorB\", 1.6),\n",
    "    (datetime.datetime(2022, 1, 1, 10, 16, 3), \"Phase\", 80.0),\n",
    "    (datetime.datetime(2022, 1, 1, 10, 16, 3), \"SensorA\", 1109.3),\n",
    "    (datetime.datetime(2022, 1, 1, 10, 16, 4), \"SensorB\", 1.5),\n",
    "    (datetime.datetime(2022, 1, 1, 10, 21, 5), \"Phase\", 20.0),\n",
    "    (datetime.datetime(2022, 1, 1, 10, 21, 6), \"SensorA\", 1110.1),\n",
    "    (datetime.datetime(2022, 1, 1, 10, 21, 7), \"SensorB\", 1.7),\n",
    "]\n",
    "\n",
    "df = spark.createDataFrame(input_data, schema)\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c7378a8e-383b-45ac-936f-e88a65d190ae",
   "metadata": {},
   "source": [
    "## Current non-streaming approach\n",
    "### Calculate \"CycleCount\" column, which increments by 1 every time Phase == 20.0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "e5835b98-13a1-4220-9422-474cdbde1b6a",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+-------+------+----------+\n",
      "|           DateTime|    Key| Value|CycleCount|\n",
      "+-------------------+-------+------+----------+\n",
      "|2022-01-01 10:01:01|  Phase|  20.0|         1|\n",
      "|2022-01-01 10:01:02|SensorA|1103.2|         1|\n",
      "|2022-01-01 10:01:02|SensorB|   1.3|         1|\n",
      "|2022-01-01 10:05:03|  Phase|  80.0|         1|\n",
      "|2022-01-01 10:05:03|SensorA|1107.7|         1|\n",
      "|2022-01-01 10:05:04|SensorB|   1.4|         1|\n",
      "|2022-01-01 10:11:05|  Phase|  20.0|         2|\n",
      "|2022-01-01 10:11:06|SensorA|1108.1|         2|\n",
      "|2022-01-01 10:11:07|SensorB|   1.6|         2|\n",
      "|2022-01-01 10:16:03|  Phase|  80.0|         2|\n",
      "|2022-01-01 10:16:03|SensorA|1109.3|         2|\n",
      "|2022-01-01 10:16:04|SensorB|   1.5|         2|\n",
      "|2022-01-01 10:21:05|  Phase|  20.0|         3|\n",
      "|2022-01-01 10:21:06|SensorA|1110.1|         3|\n",
      "|2022-01-01 10:21:07|SensorB|   1.7|         3|\n",
      "+-------------------+-------+------+----------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "24/03/04 21:30:34 WARN WindowExec: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.\n",
      "24/03/04 21:30:34 WARN WindowExec: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.\n",
      "24/03/04 21:30:34 WARN WindowExec: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.\n",
      "24/03/04 21:30:34 WARN WindowExec: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.\n",
      "24/03/04 21:30:34 WARN WindowExec: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.\n"
     ]
    }
   ],
   "source": [
    "cumsum_window = W.orderBy(\"DateTime\").rowsBetween(W.unboundedPreceding, W.currentRow)\n",
    "df = (\n",
    "    df\n",
    "    .withColumn(\"CycleCount\", F.when(F.col(\"Value\") == 20.0, F.lit(1)).otherwise(F.lit(0)))\n",
    "    .withColumn(\"CycleCount\", F.sum(F.col(\"CycleCount\")).over(cumsum_window))\n",
    ")\n",
    "df.show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "c62b5679-1b8c-40bf-97cb-5e63084869db",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+-------+------+----------+\n",
      "|           DateTime|    Key| Value|CycleCount|\n",
      "+-------------------+-------+------+----------+\n",
      "|2022-01-01 10:01:02|SensorA|1103.2|         1|\n",
      "|2022-01-01 10:01:02|SensorB|   1.3|         1|\n",
      "|2022-01-01 10:05:03|SensorA|1107.7|         1|\n",
      "|2022-01-01 10:05:04|SensorB|   1.4|         1|\n",
      "|2022-01-01 10:11:06|SensorA|1108.1|         2|\n",
      "|2022-01-01 10:11:07|SensorB|   1.6|         2|\n",
      "|2022-01-01 10:16:03|SensorA|1109.3|         2|\n",
      "|2022-01-01 10:16:04|SensorB|   1.5|         2|\n",
      "|2022-01-01 10:21:06|SensorA|1110.1|         3|\n",
      "|2022-01-01 10:21:07|SensorB|   1.7|         3|\n",
      "+-------------------+-------+------+----------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "24/03/04 21:31:19 WARN WindowExec: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.\n",
      "24/03/04 21:31:19 WARN WindowExec: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.\n",
      "24/03/04 21:31:19 WARN WindowExec: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.\n",
      "24/03/04 21:31:19 WARN WindowExec: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.\n",
      "24/03/04 21:31:19 WARN WindowExec: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.\n"
     ]
    }
   ],
   "source": [
    "df_sensors = df.filter(F.col(\"Key\").isin([\"SensorA\", \"SensorB\"]))\n",
    "df_sensors.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e088025e-588f-4175-b711-4c2299fd2b57",
   "metadata": {},
   "source": [
    "### Calculate Features"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "38f37049-b5e1-4000-a72b-3aba82d71ce5",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "24/03/04 21:33:49 WARN WindowExec: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.\n",
      "24/03/04 21:33:49 WARN WindowExec: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.\n",
      "24/03/04 21:33:49 WARN WindowExec: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.\n",
      "24/03/04 21:33:49 WARN WindowExec: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-------+------------------+-------------------+\n",
      "|CycleCount|    Key|        Value_mean|          Value_std|\n",
      "+----------+-------+------------------+-------------------+\n",
      "|         1|SensorA|           1105.45|  3.181980515339464|\n",
      "|         1|SensorB|              1.35|0.07071067811865465|\n",
      "|         2|SensorA|1108.6999999999998| 0.8485281374238892|\n",
      "|         2|SensorB|              1.55|0.07071067811865482|\n",
      "|         3|SensorA|            1110.1|               NULL|\n",
      "|         3|SensorB|               1.7|               NULL|\n",
      "+----------+-------+------------------+-------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_features = (\n",
    "    df_sensors\n",
    "    .groupBy(\"CycleCount\", \"Key\")\n",
    "    .agg(\n",
    "        F.mean(F.col(\"Value\")).alias(\"Value_mean\"),\n",
    "        F.std(F.col(\"Value\")).alias(\"Value_std\")\n",
    "    )\n",
    ")\n",
    "df_features.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "be36b02b-e828-4b9e-98d5-123add1a38ba",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.9.18"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}