Pyspark- DataframesPDF.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "c061e514",
   "metadata": {},
   "outputs": [],
   "source": [
    "conf = SparkConf(). \\\n",
    "    setMaster(\"local\"). \\\n",
    "    setAppName(\"Orders_Revenue\"). \\\n",
    "    set(\"conf.ui.port\",\"10567\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "f448b24e",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark import SparkConf, SparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "cf9e651d",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Warning: Ignoring non-Spark config property: conf.ui.port\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/11/21 15:25:33 WARN Utils: Your hostname, codeStation resolves to a loopback address: 127.0.1.1; using 192.168.160.83 instead (on interface wlo1)\n",
      "22/11/21 15:25:33 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/11/21 15:25:35 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n"
     ]
    }
   ],
   "source": [
    "sc = SparkContext(conf=conf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "b0b61326",
   "metadata": {},
   "outputs": [],
   "source": [
    "productPath = \"/home/solverbot/spark-warehouse/retail_db/products/part-00000\"\n",
    "orderitemPath = \"/home/solverbot/spark-warehouse/retail_db/order_items/part-00000\"\n",
    "ordersPath = \"/home/solverbot/spark-warehouse/retail_db/orders/part-00000.txt\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "f609fb46",
   "metadata": {},
   "outputs": [],
   "source": [
    "#What is the difference between Session and Context?\n",
    "spark = SparkSession.builder.appName('newSession').getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "92622c04",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(order_item_id='1', order_item_order_id='1', product_id='957', qty='1', product_cost='299.98', order_subtotal='299.98'),\n",
       " Row(order_item_id='2', order_item_order_id='2', product_id='1073', qty='1', product_cost='199.99', order_subtotal='199.99')]"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "orderDF = spark.read.csv(\"/home/solverbot/spark-warehouse/retail_db/order_items/\") \\\n",
    "                    .toDF(\"order_item_id\",\"order_item_order_id\",\"product_id\", \"qty\",\"product_cost\",\"order_subtotal\")\n",
    "orderDF.head(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "25050fca",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.sql.dataframe.DataFrame"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(orderDF)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "885564ce",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'1,1,957,1,299.98,299.98'"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "open(orderitemPath).read().splitlines()[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "8f3db55d",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "['1,1,957,1,299.98,299.98', '2,2,1073,1,199.99,199.99']"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc.textFile(orderitemPath).take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "c4d25568",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- order_item_id: string (nullable = true)\n",
      " |-- order_item_order_id: string (nullable = true)\n",
      " |-- product_id: string (nullable = true)\n",
      " |-- qty: string (nullable = true)\n",
      " |-- product_cost: string (nullable = true)\n",
      " |-- order_subtotal: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Note, even the numbers are read as strings when the DF is created\n",
    "orderDF.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "f3b5bd1b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+--------------+\n",
      "|order_item_order_id|order_subtotal|\n",
      "+-------------------+--------------+\n",
      "|                  1|        299.98|\n",
      "|                  2|        199.99|\n",
      "|                  2|          50.0|\n",
      "|                  2|        129.99|\n",
      "|                  4|         24.99|\n",
      "|                  4|         59.99|\n",
      "|                  4|          50.0|\n",
      "|                  4|         49.98|\n",
      "|                  5|        299.98|\n",
      "|                  5|         59.99|\n",
      "|                  5|         49.98|\n",
      "|                  5|        299.98|\n",
      "|                  5|        129.99|\n",
      "|                  7|        199.99|\n",
      "|                  7|        299.98|\n",
      "|                  7|         15.99|\n",
      "|                  8|         59.99|\n",
      "|                  8|         59.99|\n",
      "|                  8|         49.98|\n",
      "|                  8|          50.0|\n",
      "+-------------------+--------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "orderDF.select(\"order_item_order_id\",\"order_subtotal\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7d5f5005",
   "metadata": {},
   "outputs": [],
   "source": [
    "orderDF.describe().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e2f5186e",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "#Interesting option to convert DF to SQL table\n",
    "\n",
    "orderDF.createTempView(\"ordersView\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "47044ac0",
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.sql(\"SELECT * FROM ordersView LIMIT 5\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "e436f108",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://172.17.0.1:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v3.3.1</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>Orders_Revenue</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        "
      ],
      "text/plain": [
       "<SparkContext master=local appName=Orders_Revenue>"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#The sparksession contains the spark context. Using the below command the context can be \n",
    "#invoked, and the details of the runs can be checked\n",
    "spark.sparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "f71a2ed6",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "orderDFInfer = spark.read.csv(\"/home/solverbot/spark-warehouse/retail_db/order_items/\",inferSchema=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "5bf95b46",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- _c0: integer (nullable = true)\n",
      " |-- _c1: integer (nullable = true)\n",
      " |-- _c2: integer (nullable = true)\n",
      " |-- _c3: integer (nullable = true)\n",
      " |-- _c4: double (nullable = true)\n",
      " |-- _c5: double (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Recollected that inferSchema can be used to get the types of the columns correct\n",
    "orderDFInfer.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "65a46de7",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import IntegerType\n",
    "\n",
    "orderDF = orderDF \\\n",
    "            .withColumn('order_item_id',orderDF.order_item_id.cast(IntegerType())) \\\n",
    "            .withColumn('order_item_order_id',orderDF.order_item_order_id.cast(IntegerType()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "408821a1",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- order_item_id: integer (nullable = true)\n",
      " |-- order_item_order_id: integer (nullable = true)\n",
      " |-- product_id: string (nullable = true)\n",
      " |-- qty: string (nullable = true)\n",
      " |-- product_cost: string (nullable = true)\n",
      " |-- order_subtotal: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "orderDF.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8cd2cf54",
   "metadata": {},
   "source": [
    "### Integrating the database connectivity using jdbc connectors"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "092a3b60",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/11/21 16:23:15 WARN Utils: Your hostname, codeStation resolves to a loopback address: 127.0.1.1; using 192.168.160.83 instead (on interface wlo1)\n",
      "22/11/21 16:23:15 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address\n",
      "22/11/21 16:23:17 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n"
     ]
    }
   ],
   "source": [
    "sparkSQL = SparkSession.builder \\\n",
    "       .appName(\"Python Spark SQL basic example\") \\\n",
    "       .config(\"spark.jars\", \"/usr/share/java/postgresql-42.2.26.jar\") \\\n",
    "       .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "72c17044",
   "metadata": {},
   "outputs": [],
   "source": [
    "orders_df = sparkSQL.read \\\n",
    "    .format(\"jdbc\") \\\n",
    "    .option(\"url\", \"jdbc:postgresql://localhost:5432/postgres\") \\\n",
    "    .option(\"dbtable\", \"orders\") \\\n",
    "    .option(\"user\", \"postgres\") \\\n",
    "    .option(\"password\", 1234) \\\n",
    "    .option(\"driver\", \"org.postgresql.Driver\") \\\n",
    "    .load()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "dacc6683",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.sql.dataframe.DataFrame"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(orders_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "5f072673",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(order_id=1, order_date=datetime.datetime(2013, 7, 25, 0, 0), order_customer_id=11599, order_status='CLOSED'),\n",
       " Row(order_id=2, order_date=datetime.datetime(2013, 7, 25, 0, 0), order_customer_id=256, order_status='PENDING_PAYMENT')]"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "orders_df.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "201e231a",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- order_id: integer (nullable = true)\n",
      " |-- order_date: timestamp (nullable = true)\n",
      " |-- order_customer_id: integer (nullable = true)\n",
      " |-- order_status: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "orders_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "007eeec8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-------------------+-----------------+---------------+\n",
      "|order_id|         order_date|order_customer_id|   order_status|\n",
      "+--------+-------------------+-----------------+---------------+\n",
      "|       1|2013-07-25 00:00:00|            11599|         CLOSED|\n",
      "|       2|2013-07-25 00:00:00|              256|PENDING_PAYMENT|\n",
      "+--------+-------------------+-----------------+---------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "orders_df.show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "id": "63ab5703",
   "metadata": {},
   "outputs": [],
   "source": [
    "url = \"jdbc:postgresql://localhost/\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "823daa4e",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Help on method jdbc in module pyspark.sql.readwriter:\n",
      "\n",
      "jdbc(url: str, table: str, column: Optional[str] = None, lowerBound: Union[str, int, NoneType] = None, upperBound: Union[str, int, NoneType] = None, numPartitions: Optional[int] = None, predicates: Optional[List[str]] = None, properties: Optional[Dict[str, str]] = None) -> 'DataFrame' method of pyspark.sql.readwriter.DataFrameReader instance\n",
      "    Construct a :class:`DataFrame` representing the database table named ``table``\n",
      "    accessible via JDBC URL ``url`` and connection ``properties``.\n",
      "    \n",
      "    Partitions of the table will be retrieved in parallel if either ``column`` or\n",
      "    ``predicates`` is specified. ``lowerBound``, ``upperBound`` and ``numPartitions``\n",
      "    is needed when ``column`` is specified.\n",
      "    \n",
      "    If both ``column`` and ``predicates`` are specified, ``column`` will be used.\n",
      "    \n",
      "    .. versionadded:: 1.4.0\n",
      "    \n",
      "    Parameters\n",
      "    ----------\n",
      "    table : str\n",
      "        the name of the table\n",
      "    column : str, optional\n",
      "        alias of ``partitionColumn`` option. Refer to ``partitionColumn`` in\n",
      "        `Data Source Option <https://spark.apache.org/docs/latest/sql-data-sources-jdbc.html#data-source-option>`_\n",
      "        in the version you use.\n",
      "    predicates : list, optional\n",
      "        a list of expressions suitable for inclusion in WHERE clauses;\n",
      "        each one defines one partition of the :class:`DataFrame`\n",
      "    properties : dict, optional\n",
      "        a dictionary of JDBC database connection arguments. Normally at\n",
      "        least properties \"user\" and \"password\" with their corresponding values.\n",
      "        For example { 'user' : 'SYSTEM', 'password' : 'mypassword' }\n",
      "    \n",
      "    Other Parameters\n",
      "    ----------------\n",
      "    Extra options\n",
      "        For the extra options, refer to\n",
      "        `Data Source Option <https://spark.apache.org/docs/latest/sql-data-sources-jdbc.html#data-source-option>`_\n",
      "        in the version you use.\n",
      "    \n",
      "        .. # noqa\n",
      "    \n",
      "    Notes\n",
      "    -----\n",
      "    Don't create too many partitions in parallel on a large cluster;\n",
      "    otherwise Spark might crash your external database systems.\n",
      "    \n",
      "    Returns\n",
      "    -------\n",
      "    :class:`DataFrame`\n",
      "\n"
     ]
    }
   ],
   "source": [
    "help(sparkSQL.read.jdbc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "id": "4fcf2e5c",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "orders_df.write.csv('ordersdb')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "id": "776d8687",
   "metadata": {},
   "outputs": [],
   "source": [
    "#Remember to give the alias to the table that is created with select\n",
    "ordersQuery = sparkSQL.read.format('jdbc') \\\n",
    "        .option('url',\"jdbc:postgresql://localhost/postgres\") \\\n",
    "        .option('dbtable','(SELECT * FROM orders LIMIT 5) q') \\\n",
    "        .option('user','postgres') \\\n",
    "        .option('password',1234) \\\n",
    "        .option(\"driver\", \"org.postgresql.Driver\") \\\n",
    "        .load()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "id": "92a99a76",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-------------------+-----------------+---------------+\n",
      "|order_id|         order_date|order_customer_id|   order_status|\n",
      "+--------+-------------------+-----------------+---------------+\n",
      "|       1|2013-07-25 00:00:00|            11599|         CLOSED|\n",
      "|       2|2013-07-25 00:00:00|              256|PENDING_PAYMENT|\n",
      "|       3|2013-07-25 00:00:00|            12111|       COMPLETE|\n",
      "|       4|2013-07-25 00:00:00|             8827|         CLOSED|\n",
      "|       5|2013-07-25 00:00:00|            11318|       COMPLETE|\n",
      "+--------+-------------------+-----------------+---------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ordersQuery.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "id": "43e83bf7",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark.sql.functions as sf"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "id": "8b861ad3",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "id": "8b244373",
   "metadata": {},
   "outputs": [],
   "source": [
    "orderitems= sparkSQL.read.format('jdbc') \\\n",
    "        .option('url',\"jdbc:postgresql://localhost/postgres\") \\\n",
    "        .option('dbtable','order_items') \\\n",
    "        .option('user','postgres') \\\n",
    "        .option('password',1234) \\\n",
    "        .option(\"driver\", \"org.postgresql.Driver\") \\\n",
    "        .load()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "id": "4c8d6122",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- order_item_id: integer (nullable = true)\n",
      " |-- order_item_order_id: integer (nullable = true)\n",
      " |-- order_item_product_id: integer (nullable = true)\n",
      " |-- order_item_quantity: integer (nullable = true)\n",
      " |-- order_item_subtotal: double (nullable = true)\n",
      " |-- order_item_product_price: double (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "orderitems.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "id": "6ce4afec",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+-------------------+\n",
      "|order_item_quantity|order_item_subtotal|\n",
      "+-------------------+-------------------+\n",
      "|                  1|             299.98|\n",
      "|                  1|             199.99|\n",
      "|                  5|              250.0|\n",
      "|                  1|             129.99|\n",
      "|                  2|              49.98|\n",
      "|                  5|             299.95|\n",
      "|                  3|              150.0|\n",
      "|                  4|             199.92|\n",
      "|                  1|             299.98|\n",
      "|                  5|             299.95|\n",
      "|                  2|              99.96|\n",
      "|                  1|             299.98|\n",
      "|                  1|             129.99|\n",
      "|                  1|             199.99|\n",
      "|                  1|             299.98|\n",
      "|                  5|              79.95|\n",
      "|                  3|             179.97|\n",
      "|                  5|             299.95|\n",
      "|                  4|             199.92|\n",
      "|                  1|               50.0|\n",
      "+-------------------+-------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "orderitems.select(\"order_item_quantity\", \"order_item_subtotal\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "id": "7c697bde",
   "metadata": {},
   "outputs": [],
   "source": [
    "productTable = sparkSQL.read.format('jdbc') \\\n",
    "        .option('url',\"jdbc:postgresql://localhost/postgres\") \\\n",
    "        .option('dbtable','products') \\\n",
    "        .option('user','postgres') \\\n",
    "        .option('password',1234) \\\n",
    "        .option(\"driver\", \"org.postgresql.Driver\") \\\n",
    "        .load()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "id": "f12907ef",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-------------------+--------------------+-------------------+-------------+--------------------+\n",
      "|product_id|product_category_id|        product_name|product_description|product_price|       product_image|\n",
      "+----------+-------------------+--------------------+-------------------+-------------+--------------------+\n",
      "|         1|                  2|Quest Q64 10 FT. ...|                   |        59.98|http://images.acm...|\n",
      "|         2|                  2|Under Armour Men'...|                   |       129.99|http://images.acm...|\n",
      "+----------+-------------------+--------------------+-------------------+-------------+--------------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "productTable.show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "id": "dd77963f",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------------------------------------------------------------------+\n",
      "|CASE WHEN (order_status IN (COMPLETE, CLOSED)) THEN COMPLETED ELSE PENDING END|\n",
      "+------------------------------------------------------------------------------+\n",
      "|                                                                     COMPLETED|\n",
      "|                                                                       PENDING|\n",
      "+------------------------------------------------------------------------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "orders_df.selectExpr(\"CASE WHEN order_status IN ('COMPLETE','CLOSED') THEN 'COMPLETED' ELSE 'PENDING' END\").show(2)"
   ]
  }
 ],
 "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.10.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}