4. DF join operations .ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql import Row\n",
    "import collections\n",
    "from pyspark.sql.functions import col"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_list = [('2021-01-01','Test1','2021-01-01','B'),('2021-04-01','Test2','2021-03-31','B'),('2021-03-01','Test3','2021-02-28','F'),\n",
    "           ('2021-02-01','Test4','2021-29-01','N'),('2021-01-01','Test5','2021-01-01','F'),('2021-02-01','Test6','2021-01-31','B')]\n",
    "cols = ['NUDGE_END','NUDGE_ID','AS_OF_DT','NUDGE_TYPE']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.appName(\"DF-Ops\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = spark.createDataFrame(df_list,cols)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- NUDGE_END: string (nullable = true)\n",
      " |-- NUDGE_ID: string (nullable = true)\n",
      " |-- AS_OF_DT: string (nullable = true)\n",
      " |-- NUDGE_TYPE: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "data.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = data.withColumn(\"NUDGE_END\", (col(\"NUDGE_END\").cast(\"date\"))).withColumn(\"AS_OF_DT\", (col(\"AS_OF_DT\").cast(\"date\"))).where(col('NUDGE_END') == col('AS_OF_DT') + 1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------+----------+----------+\n",
      "| NUDGE_END|NUDGE_ID|  AS_OF_DT|NUDGE_TYPE|\n",
      "+----------+--------+----------+----------+\n",
      "|2021-04-01|   Test2|2021-03-31|         B|\n",
      "|2021-03-01|   Test3|2021-02-28|         F|\n",
      "|2021-02-01|   Test6|2021-01-31|         B|\n",
      "+----------+--------+----------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------+----------+----------+\n",
      "| NUDGE_END|NUDGE_ID|  AS_OF_DT|NUDGE_TYPE|\n",
      "+----------+--------+----------+----------+\n",
      "|2021-04-01|   Test2|2021-03-31|         B|\n",
      "|2021-02-01|   Test6|2021-01-31|         B|\n",
      "+----------+--------+----------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "final = df.where(col('NUDGE_TYPE')!='F')\n",
    "final.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "final.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------+----------+----------+\n",
      "| NUDGE_END|NUDGE_ID|  AS_OF_DT|NUDGE_TYPE|\n",
      "+----------+--------+----------+----------+\n",
      "|2021-04-01|   Test2|2021-03-31|         B|\n",
      "+----------+--------+----------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "final1 = final.where(col(\"AS_OF_DT\") == '2021-03-31')\n",
    "final1.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------+----------+----------+-------------------+\n",
      "| NUDGE_END|NUDGE_ID|  AS_OF_DT|NUDGE_TYPE|      LST_ACT_BY_TM|\n",
      "+----------+--------+----------+----------+-------------------+\n",
      "|2021-01-01|   Test1|2021-01-01|         B|2021-01-01 11:00:00|\n",
      "|2021-04-01|   Test3|2021-03-31|         B|2021-01-01 11:00:00|\n",
      "|2021-03-01|   Test4|2021-02-28|         F|2021-01-01 11:00:00|\n",
      "|2021-02-01|   Test1|2021-29-01|         N|2021-01-01 11:00:00|\n",
      "|2021-01-01|   Test4|2021-01-01|         F|2021-01-01 11:00:00|\n",
      "|2021-02-01|   Test6|2021-01-31|         B|2021-01-01 11:00:00|\n",
      "+----------+--------+----------+----------+-------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df1_list = [('2021-01-01','Test1','2021-01-01','B','2021-01-01 11:00:00'),('2021-04-01','Test3','2021-03-31','B','2021-01-01 11:00:00'),('2021-03-01','Test4','2021-02-28','F','2021-01-01 11:00:00'),\n",
    "           ('2021-02-01','Test1','2021-29-01','N','2021-01-01 11:00:00'),('2021-01-01','Test4','2021-01-01','F','2021-01-01 11:00:00'),('2021-02-01','Test6','2021-01-31','B','2021-01-01 11:00:00')]\n",
    "cols1 = ['NUDGE_END','NUDGE_ID','AS_OF_DT','NUDGE_TYPE','LST_ACT_BY_TM']\n",
    "fdbk = spark.createDataFrame(df1_list,cols1)\n",
    "fdbk.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------+----------+----------+-------------------+-------------+\n",
      "| NUDGE_END|NUDGE_ID|  AS_OF_DT|NUDGE_TYPE|      LST_ACT_BY_TM|LST_ACT_BY_DT|\n",
      "+----------+--------+----------+----------+-------------------+-------------+\n",
      "|2021-01-01|   Test1|2021-01-01|         B|2021-01-01 11:00:00|   2021-01-01|\n",
      "|2021-04-01|   Test3|2021-03-31|         B|2021-01-01 11:00:00|   2021-01-01|\n",
      "|2021-03-01|   Test4|2021-02-28|         F|2021-01-01 11:00:00|   2021-01-01|\n",
      "|2021-02-01|   Test1|2021-29-01|         N|2021-01-01 11:00:00|   2021-01-01|\n",
      "|2021-01-01|   Test4|2021-01-01|         F|2021-01-01 11:00:00|   2021-01-01|\n",
      "|2021-02-01|   Test6|2021-01-31|         B|2021-01-01 11:00:00|   2021-01-01|\n",
      "+----------+--------+----------+----------+-------------------+-------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "fdbk = fdbk.withColumn(\"LST_ACT_BY_DT\", (col(\"LST_ACT_BY_TM\").cast(\"date\")))\n",
    "fdbk.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------+----------+----------+----------+--------+----------+----------+-------------------+-------------+\n",
      "| NUDGE_END|NUDGE_ID|  AS_OF_DT|NUDGE_TYPE| NUDGE_END|NUDGE_ID|  AS_OF_DT|NUDGE_TYPE|      LST_ACT_BY_TM|LST_ACT_BY_DT|\n",
      "+----------+--------+----------+----------+----------+--------+----------+----------+-------------------+-------------+\n",
      "|2021-04-01|   Test2|2021-03-31|         B|      null|    null|      null|      null|               null|         null|\n",
      "|2021-02-01|   Test6|2021-01-31|         B|2021-02-01|   Test6|2021-01-31|         B|2021-01-01 11:00:00|   2021-01-01|\n",
      "|2021-01-01|   Test5|2021-01-01|         F|      null|    null|      null|      null|               null|         null|\n",
      "|2021-01-01|   Test1|2021-01-01|         B|2021-01-01|   Test1|2021-01-01|         B|2021-01-01 11:00:00|   2021-01-01|\n",
      "|2021-01-01|   Test1|2021-01-01|         B|2021-02-01|   Test1|2021-29-01|         N|2021-01-01 11:00:00|   2021-01-01|\n",
      "|2021-03-01|   Test3|2021-02-28|         F|2021-04-01|   Test3|2021-03-31|         B|2021-01-01 11:00:00|   2021-01-01|\n",
      "|2021-02-01|   Test4|2021-29-01|         N|2021-03-01|   Test4|2021-02-28|         F|2021-01-01 11:00:00|   2021-01-01|\n",
      "|2021-02-01|   Test4|2021-29-01|         N|2021-01-01|   Test4|2021-01-01|         F|2021-01-01 11:00:00|   2021-01-01|\n",
      "+----------+--------+----------+----------+----------+--------+----------+----------+-------------------+-------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "result = data.join(fdbk,data['NUDGE_ID']==fdbk['NUDGE_ID'],'left')\n",
    "result.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------+----------+----------+---------+--------+--------+----------+-------------+-------------+\n",
      "| NUDGE_END|NUDGE_ID|  AS_OF_DT|NUDGE_TYPE|NUDGE_END|NUDGE_ID|AS_OF_DT|NUDGE_TYPE|LST_ACT_BY_TM|LST_ACT_BY_DT|\n",
      "+----------+--------+----------+----------+---------+--------+--------+----------+-------------+-------------+\n",
      "|2021-04-01|   Test2|2021-03-31|         B|     null|    null|    null|      null|         null|         null|\n",
      "|2021-01-01|   Test5|2021-01-01|         F|     null|    null|    null|      null|         null|         null|\n",
      "+----------+--------+----------+----------+---------+--------+--------+----------+-------------+-------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "result1 = result.where(result['LST_ACT_BY_DT'].isNull())\n",
    "result1.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------+----------+----------+------------+-------+\n",
      "| NUDGE_END|NUDGE_ID|  AS_OF_DT|NUDGE_TYPE|BUSINESSDATE|COUNTRY|\n",
      "+----------+--------+----------+----------+------------+-------+\n",
      "|2021-01-01|   Test1|2021-01-01|         B|  2021-01-01|     SG|\n",
      "|2021-04-01|   Test2|2021-03-31|         B|  2021-01-01|     SG|\n",
      "|2021-03-01|   Test3|2021-02-28|         F|  2021-01-01|     SG|\n",
      "|2021-02-01|   Test4|2021-29-01|         N|  2021-01-01|     SG|\n",
      "|2021-01-01|   Test5|2021-01-01|         F|  2021-01-01|     SG|\n",
      "|2021-02-01|   Test6|2021-01-31|         B|  2021-01-01|     SG|\n",
      "+----------+--------+----------+----------+------------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import lit\n",
    "datamodified = data.withColumn('BUSINESSDATE',lit('2021-01-01')).withColumn('COUNTRY',lit('SG'))\n",
    "datamodified.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Dropped column NUDGE_ID\n",
      "Dropped column BUSINESSDATE\n"
     ]
    }
   ],
   "source": [
    "columns = ['NUDGE_ID','BUSINESSDATE']\n",
    "for column in columns:\n",
    "    datamodified = datamodified.drop(column)\n",
    "    print(\"Dropped column \" + column)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+----------+----------+-------+\n",
      "| NUDGE_END|  AS_OF_DT|NUDGE_TYPE|COUNTRY|\n",
      "+----------+----------+----------+-------+\n",
      "|2021-01-01|2021-01-01|         B|     SG|\n",
      "|2021-04-01|2021-03-31|         B|     SG|\n",
      "|2021-03-01|2021-02-28|         F|     SG|\n",
      "|2021-02-01|2021-29-01|         N|     SG|\n",
      "|2021-01-01|2021-01-01|         F|     SG|\n",
      "|2021-02-01|2021-01-31|         B|     SG|\n",
      "+----------+----------+----------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "datamodified.show()"
   ]
  }
 ],
 "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.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}