Distributed Variables.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "eb1f7038-2e32-408f-ac14-600162264a0d",
   "metadata": {},
   "outputs": [],
   "source": [
    "myCollection = \"My Name is Darshil and I love Spark\"\\\n",
    ".split(\" \")\n",
    "\n",
    "words = spark.sparkContext.parallelize(myCollection, 2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "3377fcfc-36f2-4230-aadf-7aa04097692e",
   "metadata": {},
   "outputs": [],
   "source": [
    "supplementalData = {\"Name\":1000, \"Darshil\":200,\n",
    "                      \"love\":-300, \"Spark\":100}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "f30ab6fb-0b2d-4d99-bb51-1d035c1871af",
   "metadata": {},
   "outputs": [],
   "source": [
    "suppBroadcast = spark.sparkContext.broadcast(supplementalData)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "b03358a6-55ce-4472-8c8f-4f5d68f9913e",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'Name': 1000, 'Darshil': 200, 'love': -300, 'Spark': 100}"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "suppBroadcast.value"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "3abef626-9de9-4fc2-9484-3f8e8ac33f48",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('love', -300),\n",
       " ('My', 0),\n",
       " ('is', 0),\n",
       " ('and', 0),\n",
       " ('I', 0),\n",
       " ('Spark', 100),\n",
       " ('Darshil', 200),\n",
       " ('Name', 1000)]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "words.map(lambda word: (word, suppBroadcast.value.get(word, 0)))\\\n",
    "    .sortBy(lambda wordPair: wordPair[1])\\\n",
    "    .collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "7ddcebd4-1c19-43d7-ba90-cd68761214c5",
   "metadata": {},
   "outputs": [],
   "source": [
    "flights = spark.read.parquet(\"data/flight-data/parquet/2010-summary.parquet\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "54a006a6-d05e-40c4-917a-ff13bf4bd21f",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(DEST_COUNTRY_NAME='United States', ORIGIN_COUNTRY_NAME='Romania', count=1),\n",
       " Row(DEST_COUNTRY_NAME='United States', ORIGIN_COUNTRY_NAME='Ireland', count=264),\n",
       " Row(DEST_COUNTRY_NAME='United States', ORIGIN_COUNTRY_NAME='India', count=69),\n",
       " Row(DEST_COUNTRY_NAME='Egypt', ORIGIN_COUNTRY_NAME='United States', count=24),\n",
       " Row(DEST_COUNTRY_NAME='Equatorial Guinea', ORIGIN_COUNTRY_NAME='United States', count=1)]"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "flights.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "a5ed45fc-5ab4-4001-a52b-d06a21e791de",
   "metadata": {},
   "outputs": [],
   "source": [
    "accChina = spark.sparkContext.accumulator(0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "97e52519-ae3e-4e60-a58d-666211376b70",
   "metadata": {},
   "outputs": [],
   "source": [
    "def accChinaFunc(flight_row):\n",
    "    destination = flight_row[\"DEST_COUNTRY_NAME\"]\n",
    "    origin = flight_row[\"ORIGIN_COUNTRY_NAME\"]\n",
    "    \n",
    "    if destination == \"China\":\n",
    "      accChina.add(flight_row[\"count\"])\n",
    "        \n",
    "    if origin == \"China\":\n",
    "      accChina.add(flight_row[\"count\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "2e466874-4444-474c-b021-f176e96b4119",
   "metadata": {},
   "outputs": [],
   "source": [
    "flights.foreach(lambda flight_row: accChinaFunc(flight_row))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "39894f17-4cff-4cc5-9b86-88f1fc4fc96b",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "953"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "accChina.value"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c600f0a8-f9bb-4289-8698-19536f3e64bb",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.12.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}