day-006_the_date_and_time_challenge.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Day 6 - Date and Time Challenges\n",
    "Working with date and time data is always a struggle, because there are many different notations and formats as well as the timezone issue. So today I want to check, how the `pysparl.sql.functions` module can help me here."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark\n",
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession\\\n",
    "   .builder\\\n",
    "   .getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "There are just two kind of objects regarding date and time known to Spark\n",
    "1. `DateType`: which refers to a calender\n",
    "1. `TimeStampType`: which is a date extended by a time releated to a timezone\n",
    "\n",
    "By default, Spark derives the current date and timestamp from the local machine settings, unless it get's overruled by `SparkConf` settings."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+----------+-----------------------+\n",
      "|id |Today     |Now                    |\n",
      "+---+----------+-----------------------+\n",
      "|0  |2020-04-16|2020-04-16 12:47:08.977|\n",
      "+---+----------+-----------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "import pyspark.sql.functions\n",
    "\n",
    "df = spark\\\n",
    "    .range(1)\\\n",
    "    .withColumn(\"Today\", current_date())\\\n",
    "    .withColumn(\"Now\", current_timestamp())\n",
    "\n",
    "df.show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Decomposing Date and Timestamp Objects\n",
    "If I need to check for a particular part of a date/timestap, maybe to get all data records created in Apr-2020, I can decompose these objects by one of the following functions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-RECORD 0-------------------------------\n",
      " now          | 2020-04-16 13:54:04.238 \n",
      " year         | 2020                    \n",
      " quarter      | 2                       \n",
      " month        | 4                       \n",
      " day of month | 16                      \n",
      " hours        | 13                      \n",
      " minutes      | 54                      \n",
      " seconds      | 54                      \n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(\n",
    "    \"now\",\n",
    "    year(\"Now\").alias(\"year\"),\n",
    "    quarter(\"Now\").alias(\"quarter\"),\n",
    "    month(\"now\").alias(\"month\"),\n",
    "    dayofmonth(\"now\").alias(\"day of month\"),\n",
    "    hour(\"now\").alias(\"hours\"),\n",
    "    minute(\"now\").alias(\"minutes\"),\n",
    "    minute(\"now\").alias(\"seconds\")\n",
    ").show(vertical=True, truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "I can even derive some further attributes from a given date. Especially the `last_day()` function can help me implementing month-end reports."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 112,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-RECORD 0-------------------------------\n",
      " now          | 2020-04-16 15:42:52.371 \n",
      " week of year | 16                      \n",
      " day of week  | 5                       \n",
      " month end    | 2020-04-30              \n",
      " next Sunday  | 2020-04-19              \n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(\n",
    "    \"now\",\n",
    "    weekofyear(\"now\").alias(\"week of year\"),\n",
    "    dayofweek(\"now\").alias(\"day of week\"),\n",
    "    last_day(\"now\").alias(\"month end\"),\n",
    "    next_day(\"today\", \"Sun\").alias(\"next Sunday\")\n",
    ").show(vertical=True, truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Date Shifts and Date Period Calculation\n",
    "\n",
    "Spark provides native support for calulations of time periodes on daily or monthly basis."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 113,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-RECORD 0-----------------------------------------------------------\n",
      " today                                    | 2020-04-16              \n",
      " day after tomorrow                       | 2020-04-18              \n",
      " yesterday                                | 2020-04-15              \n",
      " yesterday                                | 2020-04-15              \n",
      " same day next month                      | 2020-05-16              \n",
      " same day prev. month                     | 2020-03-16              \n",
      " now                                      | 2020-04-16 15:43:02.217 \n",
      " tomorrow at same time                    | 2020-04-17              \n",
      " yesterday at same time                   | 2020-04-15              \n",
      " next month                               | 2020-05-16              \n",
      " # of days between yesterday and tomorrow | 2                       \n",
      " # of months between two dates            | -2.5483871              \n",
      " # of months between two dates            | 2.5483871               \n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(\n",
    "    \"today\",\n",
    "    date_add(\"today\", 2).alias(\"day after tomorrow\"),\n",
    "    date_add(\"today\", -1).alias(\"yesterday\"),\n",
    "    date_sub(\"today\", 1).alias(\"yesterday\"),\n",
    "    add_months(\"today\", 1).alias(\"same day next month\"),\n",
    "    add_months(\"today\", -1).alias(\"same day prev. month\"),\n",
    "    \"now\",\n",
    "    date_add(\"now\", 1).alias(\"tomorrow at same time\"),\n",
    "    date_sub(\"now\", 1).alias(\"yesterday at same time\"),\n",
    "    add_months(\"now\", 1).alias(\"next month\"),\n",
    "    datediff(date_add(\"today\", 1), date_sub(\"today\", 1)).alias(\"# of days between yesterday and tomorrow\"),\n",
    "    months_between(\"today\", date_add(\"today\", 77)).alias(\"# of months between two dates\"),\n",
    "    months_between(\"today\", date_sub(\"today\", 77)).alias(\"# of months between two dates\")\n",
    ").show(vertical=True, truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "It's quite counter-intuitive to me that the month diff is negative when the second date is later than the first date and vice versa. The calculation is simply date1 - date2 and since earlier dates are smaller than later ones,  the negative result obvious is absolutely intuitive for mathematicians.\n",
    "\n",
    "I should keep also in mind, that when applying these funtions to timestamps, they get truncated to dates and timestamps by myself.\n",
    "\n",
    "The `date_sub()` function is redundant because `date_add()` accepts also negative day shifts."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 76,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-RECORD 0-------------------------\n",
      " today   | 2020-04-16             \n",
      " yyyy    | 2020-01-01 00:00:00    \n",
      " year    | 2020-01-01 00:00:00    \n",
      " yy      | 2020-01-01 00:00:00    \n",
      " quarter | 2020-04-01 00:00:00    \n",
      " month   | 2020-04-01 00:00:00    \n",
      " mon     | 2020-04-01 00:00:00    \n",
      " week    | 2020-04-13 00:00:00    \n",
      " mm      | 2020-04-01 00:00:00    \n",
      " now     | 2020-04-16 14:06:22.63 \n",
      " yyyy    | 2020-01-01 00:00:00    \n",
      " year    | 2020-01-01 00:00:00    \n",
      " yy      | 2020-01-01 00:00:00    \n",
      " quarter | 2020-04-01 00:00:00    \n",
      " month   | 2020-04-01 00:00:00    \n",
      " mon     | 2020-04-01 00:00:00    \n",
      " week    | 2020-04-13 00:00:00    \n",
      " mm      | 2020-04-01 00:00:00    \n",
      " day     | 2020-04-16 00:00:00    \n",
      " dd      | 2020-04-16 00:00:00    \n",
      " hour    | 2020-04-16 14:00:00    \n",
      " minute  | 2020-04-16 14:06:00    \n",
      " second  | 2020-04-16 14:06:22    \n",
      " quarter | 2020-04-01 00:00:00    \n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(\n",
    "    \"today\",\n",
    "    date_trunc(\"yyyy\", \"today\").alias(\"yyyy\"),\n",
    "    date_trunc(\"year\", \"today\").alias(\"year\"),\n",
    "    date_trunc(\"year\", \"today\").alias(\"yy\"),\n",
    "    date_trunc(\"quarter\", \"today\").alias(\"quarter\"),\n",
    "    date_trunc(\"month\", \"today\").alias(\"month\"),\n",
    "    date_trunc(\"mon\", \"today\").alias(\"mon\"),\n",
    "    date_trunc(\"week\", \"today\").alias(\"week\"),\n",
    "    date_trunc(\"mm\", \"today\").alias(\"mm\"),\n",
    "    \"now\",\n",
    "    date_trunc(\"yyyy\", \"now\").alias(\"yyyy\"),\n",
    "    date_trunc(\"year\", \"now\").alias(\"year\"),\n",
    "    date_trunc(\"year\", \"now\").alias(\"yy\"),\n",
    "    date_trunc(\"quarter\", \"now\").alias(\"quarter\"),\n",
    "    date_trunc(\"month\", \"now\").alias(\"month\"),\n",
    "    date_trunc(\"mon\", \"now\").alias(\"mon\"),\n",
    "    date_trunc(\"week\", \"now\").alias(\"week\"),\n",
    "    date_trunc(\"mm\", \"now\").alias(\"mm\"),\n",
    "    date_trunc(\"day\", \"now\").alias(\"day\"),\n",
    "    date_trunc(\"dd\", \"now\").alias(\"dd\"),\n",
    "    date_trunc(\"hour\", \"now\").alias(\"hour\"),\n",
    "    date_trunc(\"minute\", \"now\").alias(\"minute\"),\n",
    "    date_trunc(\"second\", \"now\").alias(\"second\"),\n",
    "    date_trunc(\"quarter\", \"today\").alias(\"quarter\"),\n",
    ").show(vertical=True, truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Type Conversion: String vs. Date/Timstamps\n",
    "Especially during data import or export of dates/timestamps, I often have to read from strings or write to strings. \n",
    "\n",
    "### Reading Strings:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-------------------+\n",
      "|id |ts string          |\n",
      "+---+-------------------+\n",
      "|0  |2020-04-16 14:16:23|\n",
      "+---+-------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "stringDF = spark.range(1).withColumn(\"ts string\", lit(\"2020-04-16 14:16:23\"))\n",
    "stringDF.show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "When I read from file with `option(\"inferSchema\", \"true\")` Spark can identifiy most of the common date and datetime notations and derive the correspondig column type. If the notation is unknon to Spark, it will set the column to DataTypeString. In that case I can convert these strings to dates or timestamps during a subsequent transformation. This requires a string format according to the Java <a href=\"https://docs.oracle.com/javase/tutorial/i18n/format/simpleDateFormat.html\">SimpleDateFormats</a>\n",
    "\n",
    "**Note:** Internally Spark works with Java dates and timestamps."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 88,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+-------------------+-----------------+\n",
      "|ts string          |converted to ts    |converted to date|\n",
      "+-------------------+-------------------+-----------------+\n",
      "|2020-04-16 14:16:23|2020-04-16 14:16:23|2020-04-16       |\n",
      "+-------------------+-------------------+-----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "stringDF.select(\n",
    "        \"ts string\",\n",
    "        to_timestamp(\"ts string\", \"yyyy-MM-dd HH:mm:ss\").alias(\"converted to ts\"),\n",
    "        to_date(\"ts string\", \"yyyy-MM-dd HH:mm:ss\").alias(\"converted to date\")\n",
    ").show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Writing as Strings\n",
    "To convert either a date or a timestamp into a string, again I have to specify the format according to the Java [SimpleDateFormats](https://docs.oracle.com/javase/tutorial/i18n/format/simpleDateFormat.html)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 90,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-------------------+\n",
      "|date string|ts string          |\n",
      "+-----------+-------------------+\n",
      "|16.04.2020 |16.04:2020 14:53.11|\n",
      "+-----------+-------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(\n",
    "    date_format(\"today\", \"dd.MM.yyyy\").alias(\"date string\"),\n",
    "    date_format(\"now\", \"dd.MM:yyyy HH:mm.ss\").alias(\"ts string\")\n",
    ").show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Timezone Conversion\n",
    "Processing timestamp gets even more complicated, when I need to load data from source systems, which are located in different timezones. When I want to analyse flight durations, I should consider that the timezone offsets between daparture time and arrival time. Otherwise in my dataresult a flight from Hamburg to London might take just 5 minutes.\n",
    "\n",
    "So to ensure proper time period calculations I should normalize all timestamps to a common timezone which is the *Unified Time Coordinated (UTC)*."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 101,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------------+-----------------------+-----------------------+-----------------------+\n",
      "|now                    |CET ts in UTC          |CEST ts ts in UTC      |Europe/Berlin ts in UTC|\n",
      "+-----------------------+-----------------------+-----------------------+-----------------------+\n",
      "|2020-04-16 15:25:19.675|2020-04-16 13:25:19.675|2020-04-16 13:25:19.675|2020-04-16 13:25:19.675|\n",
      "+-----------------------+-----------------------+-----------------------+-----------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "utcDF = df.select(\n",
    "    \"now\",\n",
    "    to_utc_timestamp(\"now\", \"CET\").alias(\"CET ts in UTC\"),\n",
    "    to_utc_timestamp(\"now\", \"CET\").alias(\"CEST ts ts in UTC\"),\n",
    "    to_utc_timestamp(\"now\", \"Europe/Berlin\").alias(\"Europe/Berlin ts in UTC\"),\n",
    ")\n",
    "\n",
    "utcDF.show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Obviously, Spark is smart enough to identify, that the given timestamp is in the day-light-saving period where CET = UTC+1 get's shifted to CEST = UTC+2, so in all cases Spark applies the 'Europe/Berlin' rules regardless whether I define this rule rexplicitly or I pass the CET or CEST timezone.\n",
    "\n",
    "If I need to localize UTC timestamps, I just need to apply the reverse function.\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 102,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------------+\n",
      "|local ts               |\n",
      "+-----------------------+\n",
      "|2020-04-16 15:25:32.204|\n",
      "+-----------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "utcDF.select(\n",
    "    from_utc_timestamp(\"Europe/Berlin ts in UTC\", \"Europe/Berlin\").alias(\"local ts\")\n",
    ").show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The Unix timestamp is another alternative for timestamp harmonization:\n",
    "\n",
    "**from_unixtime(timestamp, format='yyyy-MM-dd HH:mm:ss')**\n",
    "Converts the number of seconds from unix epoch (1970-01-01 00:00:00 UTC) to a string representing the timestamp of that moment in the current system time zone in the given format.\n",
    "\n",
    "\n",
    "**unix_timestamp(timestamp=None, format='yyyy-MM-dd HH:mm:ss')**\n",
    "Convert time string with given pattern (‘yyyy-MM-dd HH:mm:ss’, by default) to Unix time stamp (in seconds), using the default timezone and the default locale, return null if fail."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.8.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}