Pyspark- Accumulators_Broadcasters.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "ed0f3d88",
   "metadata": {},
   "outputs": [],
   "source": [
    "#Working on pyspark in Jupyterlab\n",
    "import pyspark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "ee49d1ae",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "93064bae",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark import SparkConf, SparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "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": 5,
   "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 09:44:18 WARN Utils: Your hostname, codeStation resolves to a loopback address: 127.0.1.1; using 192.168.213.83 instead (on interface wlo1)\n",
      "22/11/21 09:44:18 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 09:44:20 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
      "22/11/21 09:44:24 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.\n"
     ]
    }
   ],
   "source": [
    "sc = SparkContext(conf=conf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "94e777a0",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "id": "045ed2f5",
   "metadata": {},
   "source": [
    "#### Execution of the file in command prompt\n",
    "\n",
    "spark-submit --master local --deploy-mode client --conf spark.ui.port=12901 --conf spark.dynamicAllocation.anabled=false --num-executors 1 --executor-memory 512M --executor-cores 1 file_name_for_DE.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "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": 8,
   "id": "1273d61f",
   "metadata": {},
   "outputs": [],
   "source": [
    "products = sc.textFile(productPath)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "6ac4d8a9",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.rdd.RDD"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(products)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "fe97b72c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Help on method saveAsTextFile in module pyspark.rdd:\n",
      "\n",
      "saveAsTextFile(path: str, compressionCodecClass: Optional[str] = None) -> None method of pyspark.rdd.RDD instance\n",
      "    Save this RDD as a text file, using string representations of elements.\n",
      "    \n",
      "    Parameters\n",
      "    ----------\n",
      "    path : str\n",
      "        path to text file\n",
      "    compressionCodecClass : str, optional\n",
      "        fully qualified classname of the compression codec class\n",
      "        i.e. \"org.apache.hadoop.io.compress.GzipCodec\" (None by default)\n",
      "    \n",
      "    Examples\n",
      "    --------\n",
      "    >>> from tempfile import NamedTemporaryFile\n",
      "    >>> tempFile = NamedTemporaryFile(delete=True)\n",
      "    >>> tempFile.close()\n",
      "    >>> sc.parallelize(range(10)).saveAsTextFile(tempFile.name)\n",
      "    >>> from fileinput import input\n",
      "    >>> from glob import glob\n",
      "    >>> ''.join(sorted(input(glob(tempFile.name + \"/part-0000*\"))))\n",
      "    '0\\n1\\n2\\n3\\n4\\n5\\n6\\n7\\n8\\n9\\n'\n",
      "    \n",
      "    Empty lines are tolerated when saving to text files.\n",
      "    \n",
      "    >>> from tempfile import NamedTemporaryFile\n",
      "    >>> tempFile2 = NamedTemporaryFile(delete=True)\n",
      "    >>> tempFile2.close()\n",
      "    >>> sc.parallelize(['', 'foo', '', 'bar', '']).saveAsTextFile(tempFile2.name)\n",
      "    >>> ''.join(sorted(input(glob(tempFile2.name + \"/part-0000*\"))))\n",
      "    '\\n\\n\\nbar\\nfoo\\n'\n",
      "    \n",
      "    Using compressionCodecClass\n",
      "    \n",
      "    >>> from tempfile import NamedTemporaryFile\n",
      "    >>> tempFile3 = NamedTemporaryFile(delete=True)\n",
      "    >>> tempFile3.close()\n",
      "    >>> codec = \"org.apache.hadoop.io.compress.GzipCodec\"\n",
      "    >>> sc.parallelize(['foo', 'bar']).saveAsTextFile(tempFile3.name, codec)\n",
      "    >>> from fileinput import input, hook_compressed\n",
      "    >>> result = sorted(input(glob(tempFile3.name + \"/part*.gz\"), openhook=hook_compressed))\n",
      "    >>> ''.join([r.decode('utf-8') if isinstance(r, bytes) else r for r in result])\n",
      "    'bar\\nfoo\\n'\n",
      "\n"
     ]
    }
   ],
   "source": [
    "help(products.saveAsTextFile)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "536812d6",
   "metadata": {},
   "outputs": [],
   "source": [
    "import getpass\n",
    "username = getpass.getuser()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "bce7a962",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'solverbot'"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "username"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "id": "c0dcd67f",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "57431"
      ]
     },
     "execution_count": 62,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "readingBack = sc.textFile(f\"/home/{username}/orders_and_sales\")\n",
    "readingBack.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "55c94392",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<py4j.java_gateway.JavaPackage at 0x7fc4d6767100>"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#The spark context is a jvm which can be used to access the JVM related classes. \n",
    "sc._gateway.jvm.org.apache"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "4578ebc5",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "JavaObject id=o26"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#The codecs for compressing the files can be located from spark context before using it\n",
    "sc._gateway.jvm.org.apache.hadoop.io.compress.GzipCodec()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "c350034f",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'3.3.1'"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc.version"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "35dea0ab",
   "metadata": {},
   "outputs": [],
   "source": [
    "sparksession = SparkSession.builder.appName('Read_session').getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "936acf25",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "ordersJson = sparksession.read.json(\"/home/solverbot/spark-warehouse/retail_db_json/orders\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "012c76eb",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(order_customer_id=11599, order_date='2013-07-25 00:00:00.0', order_id=1, order_status='CLOSED'),\n",
       " Row(order_customer_id=256, order_date='2013-07-25 00:00:00.0', order_id=2, order_status='PENDING_PAYMENT')]"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ordersJson.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "b304ad47",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Help on method json in module pyspark.sql.readwriter:\n",
      "\n",
      "json(path: str, mode: Optional[str] = None, compression: Optional[str] = None, dateFormat: Optional[str] = None, timestampFormat: Optional[str] = None, lineSep: Optional[str] = None, encoding: Optional[str] = None, ignoreNullFields: Union[bool, str, NoneType] = None) -> None method of pyspark.sql.readwriter.DataFrameWriter instance\n",
      "    Saves the content of the :class:`DataFrame` in JSON format\n",
      "    (`JSON Lines text format or newline-delimited JSON <http://jsonlines.org/>`_) at the\n",
      "    specified path.\n",
      "    \n",
      "    .. versionadded:: 1.4.0\n",
      "    \n",
      "    Parameters\n",
      "    ----------\n",
      "    path : str\n",
      "        the path in any Hadoop supported file system\n",
      "    mode : str, optional\n",
      "        specifies the behavior of the save operation when data already exists.\n",
      "    \n",
      "        * ``append``: Append contents of this :class:`DataFrame` to existing data.\n",
      "        * ``overwrite``: Overwrite existing data.\n",
      "        * ``ignore``: Silently ignore this operation if data already exists.\n",
      "        * ``error`` or ``errorifexists`` (default case): Throw an exception if data already                 exists.\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-json.html#data-source-option>`_\n",
      "        in the version you use.\n",
      "    \n",
      "        .. # noqa\n",
      "    \n",
      "    Examples\n",
      "    --------\n",
      "    >>> df.write.json(os.path.join(tempfile.mkdtemp(), 'data'))\n",
      "\n"
     ]
    }
   ],
   "source": [
    "help(ordersJson.write.json)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "8a773047",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "['{\"order_id\":1,\"order_date\":\"2013-07-25 00:00:00.0\",\"order_customer_id\":11599,\"order_status\":\"CLOSED\"}',\n",
       " '{\"order_id\":2,\"order_date\":\"2013-07-25 00:00:00.0\",\"order_customer_id\":256,\"order_status\":\"PENDING_PAYMENT\"}']"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ordersTextJson = sc.textFile('/home/solverbot/spark-warehouse/retail_db_json/orders')\n",
    "ordersTextJson.take(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "56281b50",
   "metadata": {},
   "source": [
    "### JSON files have different load option\n",
    "\n",
    "The json files store the data in key value pairs, and dictionaries. So when the files are read from Json text files, the additional step of converting the individual rows to dictionaries has to be completed using the Json.loads() option as shown below"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "id": "f5e6033c",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'CLOSED'"
      ]
     },
     "execution_count": 39,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import json\n",
    "json.loads(ordersTextJson.take(1)[0])[\"order_status\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "id": "9227fbb8",
   "metadata": {},
   "outputs": [],
   "source": [
    "ordersJsonDict = ordersTextJson.map(lambda d: json.loads(d))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "id": "05baec31",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['{\"order_id\":1,\"order_date\":\"2013-07-25 00:00:00.0\",\"order_customer_id\":11599,\"order_status\":\"CLOSED\"}']"
      ]
     },
     "execution_count": 41,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ordersTextJson.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "id": "f12fe04f",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CLOSED\n",
      "PENDING_PAYMENT\n",
      "COMPLETE\n",
      "CLOSED\n",
      "COMPLETE\n"
     ]
    }
   ],
   "source": [
    "for i in ordersJsonDict.take(5): print(i['order_status'])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "65111028",
   "metadata": {},
   "source": [
    "### Some of the map options are very interesting"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "id": "c74fc5f4",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['1,2013-07-25 00:00:00.0,11599,CLOSED',\n",
       " '2,2013-07-25 00:00:00.0,256,PENDING_PAYMENT']"
      ]
     },
     "execution_count": 46,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "orders = sc.textFile(ordersPath)\n",
    "orders.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "id": "a866b474",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['1,1,957,1,299.98,299.98', '2,2,1073,1,199.99,199.99']"
      ]
     },
     "execution_count": 50,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "orderItems = sc.textFile(orderitemPath)\n",
    "orderItems.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "id": "f9abc6d7",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['1297,2013-08-01 00:00:00.0,11607,COMPLETE',\n",
       " '1298,2013-08-01 00:00:00.0,5105,CLOSED']"
      ]
     },
     "execution_count": 47,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#Notice the \"check\" expression is against the first element of the split \n",
    "orders.filter(lambda order: \"2013-08\" in order.split(',')[1]).take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "id": "5020e6c1",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(1297, 1), (1298, 1), (1299, 1), (1300, 1), (1301, 1)]"
      ]
     },
     "execution_count": 56,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#Notice the \"startswith\" expression is against the first element of the split \n",
    "ordersFiltered = orders.filter(lambda order: order.split(',')[1].startswith(\"2013-08\")). \\\n",
    "                    map(lambda item: (int(item.split(',')[0]), 1))\n",
    "ordersFiltered.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "id": "d459f551",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "5680"
      ]
     },
     "execution_count": 61,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ordersFiltered.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "id": "ba10747c",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(1, (957, 299.98)), (2, (1073, 199.99))]"
      ]
     },
     "execution_count": 53,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "orderItemsFiltered = orderItems.map(lambda order: (int(order.split(',')[1]), \n",
    "                                                   (int(order.split(',')[2]), float(order.split(',')[4]))))\n",
    "    \n",
    "orderItemsFiltered.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "id": "d8fb0970",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[(1300, ((191, 199.98), 1)), (1300, ((1014, 199.92), 1))]"
      ]
     },
     "execution_count": 58,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "joinOrders = orderItemsFiltered.join(ordersFiltered)\n",
    "joinOrders.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "id": "790494ce",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(1300, ((191, 199.98), 1))\n",
      "(1300, ((1014, 199.92), 1))\n",
      "(1302, ((365, 59.99), 1))\n",
      "(1302, ((502, 100.0), 1))\n",
      "(1302, ((365, 239.96), 1))\n",
      "(1306, ((191, 399.96), 1))\n",
      "(1306, ((191, 299.97), 1))\n",
      "(1306, ((1004, 399.98), 1))\n",
      "(1308, ((1073, 199.99), 1))\n",
      "(1308, ((1073, 199.99), 1))\n"
     ]
    }
   ],
   "source": [
    "for i in joinOrders.take(10): print(i)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "id": "f6fb5902",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "14268"
      ]
     },
     "execution_count": 60,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "joinOrders.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "id": "48543d13",
   "metadata": {},
   "outputs": [],
   "source": [
    "revenueByPid=joinOrders. \\\n",
    "                map(lambda rec: rec[1][0]). \\\n",
    "                reduceByKey(lambda tot, ele: round(tot + ele, 1))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "id": "9315e6da",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[(1014, 247251.2),\n",
       " (502, 261150.0),\n",
       " (1004, 562400.0),\n",
       " (804, 2039.9),\n",
       " (172, 2340.0)]"
      ]
     },
     "execution_count": 73,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "revenueByPid.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "id": "965e0d34",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['1,2,Quest Q64 10 FT. x 10 FT. Slant Leg Instant U,,59.98,http://images.acmesports.sports/Quest+Q64+10+FT.+x+10+FT.+Slant+Leg+Instant+Up+Canopy',\n",
       " \"2,2,Under Armour Men's Highlight MC Football Clea,,129.99,http://images.acmesports.sports/Under+Armour+Men%27s+Highlight+MC+Football+Cleat\"]"
      ]
     },
     "execution_count": 63,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#The file in the local file system can be read, lines can be split and the parallelized\n",
    "productslocal = open(productPath).read().splitlines()\n",
    "productsRdd = sc.parallelize(productslocal)\n",
    "productsRdd.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "id": "65ce92aa",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['1,2,Quest Q64 10 FT. x 10 FT. Slant Leg Instant U,,59.98,http://images.acmesports.sports/Quest+Q64+10+FT.+x+10+FT.+Slant+Leg+Instant+Up+Canopy',\n",
       " \"2,2,Under Armour Men's Highlight MC Football Clea,,129.99,http://images.acmesports.sports/Under+Armour+Men%27s+Highlight+MC+Football+Cleat\"]"
      ]
     },
     "execution_count": 62,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "products = sc.textFile(productPath)\n",
    "products.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 76,
   "id": "e63a078e",
   "metadata": {},
   "outputs": [],
   "source": [
    "productRevenue = products.map(lambda d: (int(d.split(',')[0]), d.split(',')[2])) \\\n",
    "        .join(revenueByPid)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "id": "4cd7294a",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Elevation Training Mask 2.0 => 1999.8',\n",
       " \"Nike Kids' Grade School KD VI Basketball Shoe => 1099.9\"]"
      ]
     },
     "execution_count": 78,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "productOutput = productRevenue.map(lambda p: f'{p[1][0]} => {str(p[1][1])}')\n",
    "productOutput.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "id": "a6157370",
   "metadata": {},
   "outputs": [],
   "source": [
    "ordersCount = sc.accumulator(0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 82,
   "id": "824317a0",
   "metadata": {},
   "outputs": [],
   "source": [
    "def getOrderTuples(rec):\n",
    "    ordersCount.add(1)\n",
    "    return (int(rec.split(',')[0]), 1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 83,
   "id": "f2176ffd",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(1297, 1), (1298, 1), (1299, 1), (1300, 1), (1301, 1)]"
      ]
     },
     "execution_count": 83,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ordersFilteredAccum = orders.filter(lambda order: order.split(',')[1].startswith(\"2013-08\")). \\\n",
    "                    map(lambda item: getOrderTuples(item))\n",
    "ordersFilteredAccum.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 84,
   "id": "27f32c5b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "5\n"
     ]
    }
   ],
   "source": [
    "print(ordersCount)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 85,
   "id": "1f0c4c8e",
   "metadata": {},
   "outputs": [],
   "source": [
    "ordersFilteredData = ordersFilteredAccum.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 86,
   "id": "af34b3ab",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Accumulator<id=0, value=5685>"
      ]
     },
     "execution_count": 86,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ordersCount #The earlier action was stored in the accumulator. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 90,
   "id": "dfabb59b",
   "metadata": {},
   "outputs": [],
   "source": [
    "joinOrdersAccum = orderItemsFiltered.join(ordersFilteredAccum)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 92,
   "id": "86f8956a",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[(1300, ((191, 199.98), 1)), (1300, ((1014, 199.92), 1))]"
      ]
     },
     "execution_count": 92,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "joinOrdersAccum.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 93,
   "id": "e45d490e",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Accumulator<id=0, value=11365>"
      ]
     },
     "execution_count": 93,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ordersCount"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 95,
   "id": "81663f8e",
   "metadata": {},
   "outputs": [],
   "source": [
    "#Understanding broadcasters\n",
    "\n",
    "productsDict = dict(map(lambda p: (int(p.split(',')[0]), p.split(',')[1]), productslocal))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 97,
   "id": "00388d8a",
   "metadata": {},
   "outputs": [],
   "source": [
    "bv = sc.broadcast(productsDict)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 102,
   "id": "67b671bc",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(24, ('Elevation Training Mask 2.0', 1999.8))]"
      ]
     },
     "execution_count": 102,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "productRevenue.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 105,
   "id": "c5375e0c",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['46', '24']"
      ]
     },
     "execution_count": 105,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "revenueByPid.map(lambda pdtRev: bv.value[pdtRev[0]]).take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 106,
   "id": "ea0db85d",
   "metadata": {},
   "outputs": [],
   "source": [
    "revenueByPdtIdBct = revenueByPid.map(lambda pdtRev: bv.value[pdtRev[0]])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 108,
   "id": "511b734f",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['46', '24']"
      ]
     },
     "execution_count": 108,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "revenueByPdtIdBct.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 109,
   "id": "5889954a",
   "metadata": {},
   "outputs": [],
   "source": [
    "ordersPart = sc.textFile(ordersPath, 4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 110,
   "id": "762f3db1",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "68883"
      ]
     },
     "execution_count": 110,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ordersPart.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 111,
   "id": "1afeeed0",
   "metadata": {},
   "outputs": [],
   "source": [
    "orderItemsPart = sc.textFile(orderitemPath,8)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 115,
   "id": "5287edcf",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "172198"
      ]
     },
     "execution_count": 115,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "orderItemsPart.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 116,
   "id": "5af9517b",
   "metadata": {},
   "outputs": [],
   "source": [
    "orderItemsPartMap = orderItemsPart.map(lambda x: (int(x.split(',')[0]),x))\n",
    "ordersPartMap = ordersPart.map(lambda y: (int(y.split(',')[0]), y))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 117,
   "id": "cb2c0d1c",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "172198"
      ]
     },
     "execution_count": 117,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "orderItemsPartMap.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 118,
   "id": "a850bd6e",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "68883"
      ]
     },
     "execution_count": 118,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ordersPartMap.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 128,
   "id": "419051a2",
   "metadata": {},
   "outputs": [],
   "source": [
    "#The number of tasks the item creates can be controlled using the partitions\n",
    "orderItemsPart = sc.textFile(orderitemPath,8)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 129,
   "id": "944acb75",
   "metadata": {},
   "outputs": [],
   "source": [
    "orderItemsPartMap = orderItemsPart.map(lambda x: (int(x.split(',')[1]),x))\n",
    "ordersPartMap = ordersPart.map(lambda y: (int(y.split(',')[0]), y))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 130,
   "id": "51fdcc06",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "172198"
      ]
     },
     "execution_count": 130,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "orderItemsPartMap.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6f2a4f74",
   "metadata": {},
   "source": [
    "The number of tasks created for the join will be 12, since order file is 4 parts and orderitems file is 12 part"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 131,
   "id": "f8488be7",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "172198"
      ]
     },
     "execution_count": 131,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "orderItemsPartMap.join(ordersPartMap).count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 132,
   "id": "42c0a126",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "#This will create 2 files as output\n",
    "orderItemsPartMap.coalesce(2).saveAsTextFile('ordersCoalesced')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 133,
   "id": "f97fad26",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "#This will create more files, 8#\n",
    "orderItemsPartMap.saveAsTextFile(\"ordersUncoalesced\")"
   ]
  }
 ],
 "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
}