04_processing_column_data/17_dealing_with_unix_timestamp.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Dealing with Unix Timestamp\n",
    "\n",
    "Let us understand how to deal with Unix Timestamp in Spark."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "tags": [
     "remove-cell"
    ]
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<iframe width=\"560\" height=\"315\" src=\"https://www.youtube.com/embed/cSXJR0zLiic?rel=0&amp;controls=1&amp;showinfo=0\" frameborder=\"0\" allowfullscreen></iframe>\n"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "%%HTML\n",
    "<iframe width=\"560\" height=\"315\" src=\"https://www.youtube.com/embed/cSXJR0zLiic?rel=0&amp;controls=1&amp;showinfo=0\" frameborder=\"0\" allowfullscreen></iframe>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* It is an integer and started from January 1st 1970 Midnight UTC.\n",
    "* Beginning time is also known as epoch and is incremented by 1 every second.\n",
    "* We can convert Unix Timestamp to regular date or timestamp and vice versa.\n",
    "* We can use `unix_timestamp` to convert regular date or timestamp to a unix timestamp value. For example `unix_timestamp(lit(\"2019-11-19 00:00:00\"))`\n",
    "* We can use `from_unixtime` to convert unix timestamp to regular date or timestamp. For example `from_unixtime(lit(1574101800))`\n",
    "* We can also pass format to both the functions."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let us start spark context for this Notebook so that we can execute the code provided. You can sign up for our [10 node state of the art cluster/labs](https://labs.itversity.com/plans) to learn Spark SQL using our unique integrated LMS."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "import getpass\n",
    "username = getpass.getuser()\n",
    "\n",
    "spark = SparkSession. \\\n",
    "    builder. \\\n",
    "    config('spark.ui.port', '0'). \\\n",
    "    config(\"spark.sql.warehouse.dir\", f\"/user/{username}/warehouse\"). \\\n",
    "    enableHiveSupport(). \\\n",
    "    appName(f'{username} | Python - Processing Column Data'). \\\n",
    "    master('yarn'). \\\n",
    "    getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If you are going to use CLIs, you can use Spark SQL using one of the 3 approaches.\n",
    "\n",
    "**Using Spark SQL**\n",
    "\n",
    "```\n",
    "spark2-sql \\\n",
    "    --master yarn \\\n",
    "    --conf spark.ui.port=0 \\\n",
    "    --conf spark.sql.warehouse.dir=/user/${USER}/warehouse\n",
    "```\n",
    "\n",
    "**Using Scala**\n",
    "\n",
    "```\n",
    "spark2-shell \\\n",
    "    --master yarn \\\n",
    "    --conf spark.ui.port=0 \\\n",
    "    --conf spark.sql.warehouse.dir=/user/${USER}/warehouse\n",
    "```\n",
    "\n",
    "**Using Pyspark**\n",
    "\n",
    "```\n",
    "pyspark2 \\\n",
    "    --master yarn \\\n",
    "    --conf spark.ui.port=0 \\\n",
    "    --conf spark.sql.warehouse.dir=/user/${USER}/warehouse\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Tasks\n",
    "\n",
    "Let us perform few tasks to understand how to deal with Unix Timestamp.\n",
    "\n",
    "*   Create a Dataframe by name datetimesDF with columns dateid, date and time."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "datetimes = [(20140228, \"2014-02-28\", \"2014-02-28 10:00:00.123\"),\n",
    "                     (20160229, \"2016-02-29\", \"2016-02-29 08:08:08.999\"),\n",
    "                     (20171031, \"2017-10-31\", \"2017-12-31 11:59:59.123\"),\n",
    "                     (20191130, \"2019-11-30\", \"2019-08-31 00:00:00.000\")\n",
    "                ]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "datetimesDF = spark.createDataFrame(datetimes).toDF(\"dateid\", \"date\", \"time\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+----------+-----------------------+\n",
      "|dateid  |date      |time                   |\n",
      "+--------+----------+-----------------------+\n",
      "|20140228|2014-02-28|2014-02-28 10:00:00.123|\n",
      "|20160229|2016-02-29|2016-02-29 08:08:08.999|\n",
      "|20171031|2017-10-31|2017-12-31 11:59:59.123|\n",
      "|20191130|2019-11-30|2019-08-31 00:00:00.000|\n",
      "+--------+----------+-----------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "datetimesDF.show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Get unix timestamp for dateid, date and time."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import unix_timestamp, col"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+----------+--------------------+------------+----------+----------+\n",
      "|  dateid|      date|                time|unix_date_id| unix_date| unix_time|\n",
      "+--------+----------+--------------------+------------+----------+----------+\n",
      "|20140228|2014-02-28|2014-02-28 10:00:...|  1393563600|1393563600|1393599600|\n",
      "|20160229|2016-02-29|2016-02-29 08:08:...|  1456722000|1456722000|1456751288|\n",
      "|20171031|2017-10-31|2017-12-31 11:59:...|  1509422400|1509422400|1514739599|\n",
      "|20191130|2019-11-30|2019-08-31 00:00:...|  1575090000|1575090000|1567224000|\n",
      "+--------+----------+--------------------+------------+----------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "datetimesDF. \\\n",
    "    withColumn(\"unix_date_id\", unix_timestamp(col(\"dateid\").cast(\"string\"), \"yyyyMMdd\")). \\\n",
    "    withColumn(\"unix_date\", unix_timestamp(\"date\", \"yyyy-MM-dd\")). \\\n",
    "    withColumn(\"unix_time\", unix_timestamp(\"time\")). \\\n",
    "    show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Create a Dataframe by name unixtimesDF with one column unixtime using 4 values. You can use the unix timestamp generated for time column in previous task."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "unixtimes = [(1393561800, ),\n",
    "             (1456713488, ),\n",
    "             (1514701799, ),\n",
    "             (1567189800, )\n",
    "            ]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "unixtimesDF = spark.createDataFrame(unixtimes).toDF(\"unixtime\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|  unixtime|\n",
      "+----------+\n",
      "|1393561800|\n",
      "|1456713488|\n",
      "|1514701799|\n",
      "|1567189800|\n",
      "+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "unixtimesDF.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- unixtime: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "unixtimesDF.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Get date in yyyyMMdd format and also complete timestamp."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import from_unixtime"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------+-------------------+\n",
      "|  unixtime|    date|               time|\n",
      "+----------+--------+-------------------+\n",
      "|1393561800|20140227|2014-02-27 23:30:00|\n",
      "|1456713488|20160228|2016-02-28 21:38:08|\n",
      "|1514701799|20171231|2017-12-31 01:29:59|\n",
      "|1567189800|20190830|2019-08-30 14:30:00|\n",
      "+----------+--------+-------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "unixtimesDF. \\\n",
    "    withColumn(\"date\", from_unixtime(\"unixtime\", \"yyyyMMdd\")). \\\n",
    "    withColumn(\"time\", from_unixtime(\"unixtime\")). \\\n",
    "    show()\n",
    "#yyyyMMdd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Pyspark 2",
   "language": "python",
   "name": "pyspark2"
  },
  "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.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}