MongoSpark.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "74fd9ae5-9ff9-403d-abb6-43ef327af06c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\u001b[33mDEPRECATION: Configuring installation scheme with distutils config files is deprecated and will no longer work in the near future. If you are using a Homebrew or Linuxbrew Python, please see discussion at https://github.com/Homebrew/homebrew-core/issues/76621\u001b[0m\u001b[33m\n",
      "\u001b[0mRequirement already satisfied: pyspark in /opt/homebrew/lib/python3.9/site-packages (3.2.1)\n",
      "Requirement already satisfied: pymongo in /opt/homebrew/lib/python3.9/site-packages (4.1.0)\n",
      "Requirement already satisfied: findspark in /opt/homebrew/lib/python3.9/site-packages (2.0.1)\n",
      "Requirement already satisfied: matplotlib in /opt/homebrew/lib/python3.9/site-packages (3.5.1)\n",
      "Requirement already satisfied: py4j==0.10.9.3 in /opt/homebrew/lib/python3.9/site-packages (from pyspark) (0.10.9.3)\n",
      "Requirement already satisfied: pillow>=6.2.0 in /opt/homebrew/lib/python3.9/site-packages (from matplotlib) (9.1.0)\n",
      "Requirement already satisfied: pyparsing>=2.2.1 in /opt/homebrew/lib/python3.9/site-packages (from matplotlib) (3.0.8)\n",
      "Requirement already satisfied: cycler>=0.10 in /opt/homebrew/lib/python3.9/site-packages (from matplotlib) (0.11.0)\n",
      "Requirement already satisfied: numpy>=1.17 in /opt/homebrew/lib/python3.9/site-packages (from matplotlib) (1.22.3)\n",
      "Requirement already satisfied: kiwisolver>=1.0.1 in /opt/homebrew/lib/python3.9/site-packages (from matplotlib) (1.4.2)\n",
      "Requirement already satisfied: fonttools>=4.22.0 in /opt/homebrew/lib/python3.9/site-packages (from matplotlib) (4.33.3)\n",
      "Requirement already satisfied: packaging>=20.0 in /opt/homebrew/lib/python3.9/site-packages (from matplotlib) (21.3)\n",
      "Requirement already satisfied: python-dateutil>=2.7 in /opt/homebrew/lib/python3.9/site-packages (from matplotlib) (2.8.2)\n",
      "Requirement already satisfied: six>=1.5 in /opt/homebrew/lib/python3.9/site-packages (from python-dateutil>=2.7->matplotlib) (1.16.0)\n",
      "\u001b[33mDEPRECATION: Configuring installation scheme with distutils config files is deprecated and will no longer work in the near future. If you are using a Homebrew or Linuxbrew Python, please see discussion at https://github.com/Homebrew/homebrew-core/issues/76621\u001b[0m\u001b[33m\n",
      "\u001b[0mNote: you may need to restart the kernel to use updated packages.\n"
     ]
    }
   ],
   "source": [
    "pip install pyspark pymongo findspark matplotlib"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "9631ff5e-49ce-41a4-9e58-4ee533921f70",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "22/04/27 17:37:13 WARN Utils: Your hostname, Miros-MacBook-Pro.local resolves to a loopback address: 127.0.0.1; using 10.9.161.107 instead (on interface en0)\n",
      "22/04/27 17:37:13 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address\n",
      "WARNING: An illegal reflective access operation has occurred\n",
      "WARNING: Illegal reflective access by org.apache.spark.unsafe.Platform (file:/opt/homebrew/lib/python3.9/site-packages/pyspark/jars/spark-unsafe_2.12-3.2.1.jar) to constructor java.nio.DirectByteBuffer(long,int)\n",
      "WARNING: Please consider reporting this to the maintainers of org.apache.spark.unsafe.Platform\n",
      "WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations\n",
      "WARNING: All illegal access operations will be denied in a future release\n",
      "Ivy Default Cache set to: /Users/miro/.ivy2/cache\n",
      "The jars for the packages stored in: /Users/miro/.ivy2/jars\n",
      "org.mongodb.spark#mongo-spark-connector_2.12 added as a dependency\n",
      ":: resolving dependencies :: org.apache.spark#spark-submit-parent-17dfe8aa-9096-4554-9b21-79e9aac8c483;1.0\n",
      "\tconfs: [default]\n",
      "\tfound org.mongodb.spark#mongo-spark-connector_2.12;3.0.1 in central\n",
      "\tfound org.mongodb#mongodb-driver-sync;4.0.5 in central\n",
      "\tfound org.mongodb#bson;4.0.5 in central\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      ":: loading settings :: url = jar:file:/opt/homebrew/lib/python3.9/site-packages/pyspark/jars/ivy-2.5.0.jar!/org/apache/ivy/core/settings/ivysettings.xml\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\tfound org.mongodb#mongodb-driver-core;4.0.5 in central\n",
      ":: resolution report :: resolve 93ms :: artifacts dl 4ms\n",
      "\t:: modules in use:\n",
      "\torg.mongodb#bson;4.0.5 from central in [default]\n",
      "\torg.mongodb#mongodb-driver-core;4.0.5 from central in [default]\n",
      "\torg.mongodb#mongodb-driver-sync;4.0.5 from central in [default]\n",
      "\torg.mongodb.spark#mongo-spark-connector_2.12;3.0.1 from central in [default]\n",
      "\t---------------------------------------------------------------------\n",
      "\t|                  |            modules            ||   artifacts   |\n",
      "\t|       conf       | number| search|dwnlded|evicted|| number|dwnlded|\n",
      "\t---------------------------------------------------------------------\n",
      "\t|      default     |   4   |   0   |   0   |   0   ||   4   |   0   |\n",
      "\t---------------------------------------------------------------------\n",
      ":: retrieving :: org.apache.spark#spark-submit-parent-17dfe8aa-9096-4554-9b21-79e9aac8c483\n",
      "\tconfs: [default]\n",
      "\t0 artifacts copied, 4 already retrieved (0kB/3ms)\n",
      "22/04/27 17:37:13 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
      "Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties\n",
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import SparkSession, DataFrame\n",
    "from pyspark.sql.functions import to_timestamp, to_date, col\n",
    "import findspark\n",
    "from pymongo import MongoClient\n",
    "import matplotlib.pyplot as plt\n",
    "\n",
    "findspark.init()\n",
    "\n",
    "# spark_session = SparkSession \\\n",
    "#     .builder \\\n",
    "#     .appName(\"myApp\") \\\n",
    "#     .config(\"spark.mongodb.read.connection.uri\", \"mongodb://127.0.0.1/UserDB.Arrests\") \\\n",
    "#     .config(\"spark.mongodb.write.connection.uri\", \"mongodb://127.0.0.1/UserDB.Arrests\") \\\n",
    "#     .config('spark.jars.packages', 'org.mongodb.spark:mongo-spark-connector:10.0.0') \\\n",
    "#     .getOrCreate()\n",
    "\n",
    "spark_session = SparkSession \\\n",
    "    .builder \\\n",
    "    .appName(\"myApp\") \\\n",
    "    .config(\"spark.mongodb.input.uri\", \"mongodb://127.0.0.1/UserDB.Arrests\") \\\n",
    "    .config(\"spark.mongodb.output.uri\", \"mongodb://127.0.0.1/UserDB.Arrests\") \\\n",
    "    .config('spark.jars.packages', 'org.mongodb.spark:mongo-spark-connector_2.12:3.0.1') \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "d746f70a-8c0c-4570-ba8c-00a4e834aba5",
   "metadata": {},
   "outputs": [],
   "source": [
    "df: DataFrame = spark_session.read.format(\"mongo\").load() # .withColumn(\"arrest_date\", col(\"arrest_date\").cast(\"string\")) #.withColumn(\"arrest_date\", to_date(\"arrest_date\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "a93f436e",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- _id: struct (nullable = true)\n",
      " |    |-- oid: string (nullable = true)\n",
      " |-- address_number: string (nullable = true)\n",
      " |-- arrest_date: timestamp (nullable = true)\n",
      " |-- case_number: string (nullable = true)\n",
      " |-- charges: string (nullable = true)\n",
      " |-- city: string (nullable = true)\n",
      " |-- council_district: string (nullable = true)\n",
      " |-- hamilton_county_census_tracts_2010: string (nullable = true)\n",
      " |-- latitude: string (nullable = true)\n",
      " |-- location: struct (nullable = true)\n",
      " |    |-- type: string (nullable = true)\n",
      " |    |-- coordinates: array (nullable = true)\n",
      " |    |    |-- element: double (containsNull = true)\n",
      " |-- longitude: string (nullable = true)\n",
      " |-- race: string (nullable = true)\n",
      " |-- sex: string (nullable = true)\n",
      " |-- street_name: string (nullable = true)\n",
      " |-- zip: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "31726aee-a108-43ee-8c29-1648cfb436e8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------+-------------------+-----------+--------------------+-----------+----------------+----------------------------------+------------------+--------------------+------------------+-----+------+-------------------+-----+\n",
      "|                 _id|address_number|        arrest_date|case_number|             charges|       city|council_district|hamilton_county_census_tracts_2010|          latitude|            location|         longitude| race|   sex|        street_name|  zip|\n",
      "+--------------------+--------------+-------------------+-----------+--------------------+-----------+----------------+----------------------------------+------------------+--------------------+------------------+-----+------+-------------------+-----+\n",
      "|{624de5bda6fb4b39...|          7200|2022-04-04 19:00:00|  22-034544|     Serving Warrant|Chattanooga|               4|                            011446|         35.041401|{Point, [-85.1552...|        -85.155236|White|  Male|     Shallowford Rd|37421|\n",
      "|{624de5bda6fb4b39...|          2300|2022-04-04 18:24:00|  22-034525|Serving Warrant, ...|Chattanooga|               7|                            002400|35.003068999999996|{Point, [-85.2857...|-85.28578200000001|Black|  Male|          E 35th St|37407|\n",
      "|{624de5bda6fb4b39...|          2000|2022-04-04 13:27:00|  22-034407|         Shoplifting|Chattanooga|               4|                            011321|         35.033654|{Point, [-85.1525...|        -85.152579|White|Female|       Gunbarrel Rd|37421|\n",
      "|{624de5bda6fb4b39...|          2600|2022-04-04 12:36:00|  22-034377|     Serving Warrant|Chattanooga|               9|                            001200|35.033716000000005|{Point, [-85.2609...|        -85.260986|White|Female|      Glenwood Pkwy|37404|\n",
      "|{624de5bda6fb4b39...|          3600|2022-04-04 06:24:00|  22-034280|           Vandalism|Chattanooga|               1|                            012100|         35.017679|{Point, [-85.3790...|        -85.379033|Black|  Male|       Cummings Hwy|37419|\n",
      "|{624de5bda6fb4b39...|          2300|2022-04-04 04:44:00|  22-034276|  Disorderly Conduct|Chattanooga|               9|                            001200|          35.03386|{Point, [-85.2676...|        -85.267665|White|Female|       Mccallie Ave|37404|\n",
      "|{624de5bda6fb4b39...|          2900|2022-04-04 04:00:00|  22-034275|Serving Warrant, ...|Chattanooga|               9|                            012300|         35.070937|{Point, [-85.2489...|         -85.24896|Black|  Male|          Taylor St|37406|\n",
      "|{624de5bda6fb4b39...|          1300|2022-04-04 01:14:00|  22-034260|Driving Under the...|Chattanooga|               2|                            000700|         35.077342|{Point, [-85.3027...|        -85.302729|White|  Male|          Dallas Rd|37405|\n",
      "|{624de5bda6fb4b39...|          3800|2022-04-03 23:48:00|  22-034233|Trespass of Real ...|Chattanooga|               7|                            001900|         35.010543|{Point, [-85.3276...|        -85.327619|White|Female|      Tennessee Ave|37409|\n",
      "|{624de5bda6fb4b39...|          5700|2022-04-03 22:36:00|  22-034221|Shoplifting, Diso...|Chattanooga|               3|                            010431|         35.153571|{Point, [-85.2456...|        -85.245617|White|Female|        Highway 153|37343|\n",
      "|{624de5bda6fb4b39...|           500|2022-04-03 22:25:00|  22-034222|Drunkenness, Tres...|Chattanooga|               7|                            003100|         35.050539|{Point, [-85.3101...|-85.31018999999999|White|  Male|          Market St|37402|\n",
      "|{624de5bda6fb4b39...|          3500|2022-04-03 21:11:00|  22-034185|Trespass of Real ...|Chattanooga|               1|                            012100|35.016352999999995|{Point, [-85.3766...|-85.37661800000001|White|  Male|       Cummings Hwy|37419|\n",
      "|{624de5bda6fb4b39...|          3500|2022-04-03 21:11:00|  22-034219|     Serving Warrant|Chattanooga|               1|                            012100|35.015345999999994|{Point, [-85.3773...|        -85.377303|White|  Male|       Cummings Hwy|37419|\n",
      "|{624de5bda6fb4b39...|          3500|2022-04-03 21:11:00|  22-034185|Trespass of Real ...|Chattanooga|               1|                            012100|         35.016047|{Point, [-85.3773...|        -85.377314|White|Female|       Cummings Hwy|37419|\n",
      "|{624de5bda6fb4b39...|          3100|2022-04-03 20:38:00|  22-034213|Serving Warrant, ...|Chattanooga|               7|                            001900|         35.018882|{Point, [-85.3212...|        -85.321284|White|Female|           Broad St|37408|\n",
      "|{624de5bda6fb4b39...|          3100|2022-04-03 20:38:00|  22-034190|         Shoplifting|Chattanooga|               7|                            001900|         35.019675|{Point, [-85.3211...|        -85.321195|White|Female|           Broad St|37408|\n",
      "|{624de5bda6fb4b39...|          2800|2022-04-03 19:06:00|  22-034175|Other Theft/Non-S...|Chattanooga|               8|                            002300|         35.014765|{Point, [-85.2931...|-85.29318599999999|White|  Male|     Rossville Blvd|37407|\n",
      "|{624de5bda6fb4b39...|          2000|2022-04-03 18:49:00|  22-034182|     Serving Warrant|Chattanooga|               4|                            011446|35.030877000000004|{Point, [-85.1549...|-85.15494899999999|Black|  Male|       Gunbarrel Rd|37421|\n",
      "|{624de5bda6fb4b39...|          2000|2022-04-03 18:49:00|  22-034158|Shoplifting, Impe...|Chattanooga|               4|                            011446|         35.031083|{Point, [-85.1553...|-85.15537099999999|Black|  Male|       Gunbarrel Rd|37421|\n",
      "|{624de5bda6fb4b39...|          null|2022-04-03 18:25:00|  22-034166|     Serving Warrant|Chattanooga|               7|                            003100|         35.045135|{Point, [-85.3086...|-85.30861300000001|Black|  Male|W 10th St/Market St|37402|\n",
      "+--------------------+--------------+-------------------+-----------+--------------------+-----------+----------------+----------------------------------+------------------+--------------------+------------------+-----+------+-------------------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "5a014425",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 2:>                                                          (0 + 1) / 1]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[Row(_id=Row(oid='624de5bda6fb4b3918b7bb01'), address_number='7200', arrest_date=datetime.datetime(2022, 4, 4, 19, 0), case_number='22-034544', charges='Serving Warrant', city='Chattanooga', council_district='4', hamilton_county_census_tracts_2010='011446', latitude='35.041401', location=Row(type='Point', coordinates=[-85.155236, 35.041401]), longitude='-85.155236', race='White', sex='Male', street_name='Shallowford Rd', zip='37421'), Row(_id=Row(oid='624de5bda6fb4b3918b7bb02'), address_number='2300', arrest_date=datetime.datetime(2022, 4, 4, 18, 24), case_number='22-034525', charges='Serving Warrant, Serving Warrant, Serving Warrant', city='Chattanooga', council_district='7', hamilton_county_census_tracts_2010='002400', latitude='35.003068999999996', location=Row(type='Point', coordinates=[-85.285782, 35.003069]), longitude='-85.28578200000001', race='Black', sex='Male', street_name='E 35th St', zip='37407')]\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "#converts DataFrame to rdd\n",
    "rdd=df.rdd\n",
    "print(rdd.take(2))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "15647054",
   "metadata": {},
   "outputs": [],
   "source": [
    "rdd.getNumPartitions()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "a7a40fe9",
   "metadata": {},
   "outputs": [],
   "source": [
    "rdd4 = rdd.repartition(4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "af2f3014",
   "metadata": {},
   "outputs": [],
   "source": [
    "rdd4.getNumPartitions()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "e76c854a-f22e-4e69-b5eb-1b1b5a7983bf",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 3:>                                                          (0 + 1) / 1]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('White', 36054), ('Black', 36088), ('Other', 430)]\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "race_pairs = rdd4.map(lambda x: (x.race, 1))\n",
    "race_count = race_pairs.reduceByKey(lambda x, y: x + y).collect()\n",
    "print(race_count)\n",
    "# plt.hist(raceCount)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "a6012e9c-0425-4bd2-a8ee-b11976ab5cce",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('Gunbarrel Rd', 3144), ('Lee Hwy', 2670), ('Brainerd Rd', 1752), ('Shallowford Rd', 1666), ('Greenway View Dr', 1644), ('Rossville Blvd', 1610), ('Market St', 1598), ('E 11th St', 1282), ('Hixson Pike', 1242), ('Cummings Hwy', 1240)]\n"
     ]
    }
   ],
   "source": [
    "street_pairs = rdd4.map(lambda x: (x.street_name, 1))\n",
    "street_count = street_pairs.reduceByKey(lambda x, y: x + y).takeOrdered(10, lambda s: -1 * s[1])\n",
    "print(street_count)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "377e4ecf-db4b-4464-850c-4bc7a2cbbbbd",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('Male', 51092), ('Female', 21478), (None, 2)]\n"
     ]
    }
   ],
   "source": [
    "gender_pairs = rdd4.map(lambda x: (x.sex, 1))\n",
    "gender_count = gender_pairs.reduceByKey(lambda x, y: x + y).takeOrdered(10, lambda s: -1 * s[1])\n",
    "print(gender_count)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4a69a203",
   "metadata": {},
   "outputs": [],
   "source": [
    "namesGrouped = race_pairs.groupByKey()\n",
    "for key, value in namesGrouped.collect():\n",
    "    print('{0}: {1}'.format(key, list(value)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b0ba0b33",
   "metadata": {},
   "outputs": [],
   "source": [
    "from operator import add\n",
    "nameCountsGrouped = namesGrouped.map(lambda x : (x[0], sum(x[1]))) \n",
    "print(nameCountsGrouped.collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6a587aa8-1281-48f1-92e7-2f1ad85a5cf0",
   "metadata": {},
   "outputs": [],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "48796602-e969-436a-b606-de48eb6a9c76",
   "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.9.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}