notebooks/gcs-search-force.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "b62d1f6b",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.conf import SparkConf\n",
    "from pyspark.context import SparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "8d65caf7",
   "metadata": {},
   "outputs": [],
   "source": [
    "conf = SparkConf() \\\n",
    "    .setMaster('local[*]') \\\n",
    "    .setAppName('test') \\\n",
    "    .set(\"spark.jars\", \"~/bin/spark-3.0.3-bin-hadoop3.2/jars/gcs-connector-hadoop3-latest.jar\") \\\n",
    "    .set(\"spark.hadoop.google.cloud.auth.service.account.enable\", \"true\") \\\n",
    "    .set(\"spark.hadoop.google.cloud.auth.service.account.json.keyfile\", \"~/.google/credentials/google_credentials.json\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "05fb4551",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING: An illegal reflective access operation has occurred\n",
      "WARNING: Illegal reflective access by org.apache.spark.unsafe.Platform (file:/home/gerry/bin/spark-3.0.3-bin-hadoop3.2/jars/spark-unsafe_2.12-3.0.3.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",
      "22/04/02 14:01:46 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",
      "22/04/02 14:01:47 WARN DependencyUtils: Local jar /home/gerry/uk_crime_data/notebooks/~/bin/spark-3.0.3-bin-hadoop3.2/jars/gcs-connector-hadoop3-latest.jar does not exist, skipping.\n",
      "22/04/02 14:01:47 INFO SparkContext: Running Spark version 3.0.3\n",
      "22/04/02 14:01:47 INFO ResourceUtils: ==============================================================\n",
      "22/04/02 14:01:47 INFO ResourceUtils: Resources for spark.driver:\n",
      "\n",
      "22/04/02 14:01:47 INFO ResourceUtils: ==============================================================\n",
      "22/04/02 14:01:47 INFO SparkContext: Submitted application: test\n",
      "22/04/02 14:01:47 INFO SecurityManager: Changing view acls to: gerry\n",
      "22/04/02 14:01:47 INFO SecurityManager: Changing modify acls to: gerry\n",
      "22/04/02 14:01:47 INFO SecurityManager: Changing view acls groups to: \n",
      "22/04/02 14:01:47 INFO SecurityManager: Changing modify acls groups to: \n",
      "22/04/02 14:01:47 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users  with view permissions: Set(gerry); groups with view permissions: Set(); users  with modify permissions: Set(gerry); groups with modify permissions: Set()\n",
      "22/04/02 14:01:47 INFO Utils: Successfully started service 'sparkDriver' on port 46223.\n",
      "22/04/02 14:01:47 INFO SparkEnv: Registering MapOutputTracker\n",
      "22/04/02 14:01:47 INFO SparkEnv: Registering BlockManagerMaster\n",
      "22/04/02 14:01:48 INFO BlockManagerMasterEndpoint: Using org.apache.spark.storage.DefaultTopologyMapper for getting topology information\n",
      "22/04/02 14:01:48 INFO BlockManagerMasterEndpoint: BlockManagerMasterEndpoint up\n",
      "22/04/02 14:01:48 INFO SparkEnv: Registering BlockManagerMasterHeartbeat\n",
      "22/04/02 14:01:48 INFO DiskBlockManager: Created local directory at /tmp/blockmgr-26e1487f-d7f6-426e-a3a0-f62902573f7e\n",
      "22/04/02 14:01:48 INFO MemoryStore: MemoryStore started with capacity 434.4 MiB\n",
      "22/04/02 14:01:48 INFO SparkEnv: Registering OutputCommitCoordinator\n",
      "22/04/02 14:01:48 INFO Utils: Successfully started service 'SparkUI' on port 4040.\n",
      "22/04/02 14:01:48 INFO SparkUI: Bound SparkUI to 0.0.0.0, and started at http://de-bootcamp.europe-west1-b.c.de-bootcamp-339509.internal:4040\n",
      "22/04/02 14:01:48 ERROR SparkContext: Failed to add ~/bin/spark-3.0.3-bin-hadoop3.2/jars/gcs-connector-hadoop3-latest.jar to Spark environment\n",
      "java.io.FileNotFoundException: Jar /home/gerry/uk_crime_data/notebooks/~/bin/spark-3.0.3-bin-hadoop3.2/jars/gcs-connector-hadoop3-latest.jar not found\n",
      "\tat org.apache.spark.SparkContext.addLocalJarFile$1(SparkContext.scala:1848)\n",
      "\tat org.apache.spark.SparkContext.addJar(SparkContext.scala:1900)\n",
      "\tat org.apache.spark.SparkContext.$anonfun$new$11(SparkContext.scala:491)\n",
      "\tat org.apache.spark.SparkContext.$anonfun$new$11$adapted(SparkContext.scala:491)\n",
      "\tat scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)\n",
      "\tat scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)\n",
      "\tat scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)\n",
      "\tat org.apache.spark.SparkContext.<init>(SparkContext.scala:491)\n",
      "\tat org.apache.spark.api.java.JavaSparkContext.<init>(JavaSparkContext.scala:58)\n",
      "\tat java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)\n",
      "\tat java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)\n",
      "\tat java.base/jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)\n",
      "\tat java.base/java.lang.reflect.Constructor.newInstance(Constructor.java:490)\n",
      "\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:247)\n",
      "\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\n",
      "\tat py4j.Gateway.invoke(Gateway.java:238)\n",
      "\tat py4j.commands.ConstructorCommand.invokeConstructor(ConstructorCommand.java:80)\n",
      "\tat py4j.commands.ConstructorCommand.execute(ConstructorCommand.java:69)\n",
      "\tat py4j.GatewayConnection.run(GatewayConnection.java:238)\n",
      "\tat java.base/java.lang.Thread.run(Thread.java:834)\n",
      "22/04/02 14:01:48 INFO Executor: Starting executor ID driver on host de-bootcamp.europe-west1-b.c.de-bootcamp-339509.internal\n",
      "22/04/02 14:01:48 INFO Utils: Successfully started service 'org.apache.spark.network.netty.NettyBlockTransferService' on port 44699.\n",
      "22/04/02 14:01:48 INFO NettyBlockTransferService: Server created on de-bootcamp.europe-west1-b.c.de-bootcamp-339509.internal:44699\n",
      "22/04/02 14:01:48 INFO BlockManager: Using org.apache.spark.storage.RandomBlockReplicationPolicy for block replication policy\n",
      "22/04/02 14:01:48 INFO BlockManagerMaster: Registering BlockManager BlockManagerId(driver, de-bootcamp.europe-west1-b.c.de-bootcamp-339509.internal, 44699, None)\n",
      "22/04/02 14:01:48 INFO BlockManagerMasterEndpoint: Registering block manager de-bootcamp.europe-west1-b.c.de-bootcamp-339509.internal:44699 with 434.4 MiB RAM, BlockManagerId(driver, de-bootcamp.europe-west1-b.c.de-bootcamp-339509.internal, 44699, None)\n",
      "22/04/02 14:01:48 INFO BlockManagerMaster: Registered BlockManager BlockManagerId(driver, de-bootcamp.europe-west1-b.c.de-bootcamp-339509.internal, 44699, None)\n",
      "22/04/02 14:01:48 INFO BlockManager: Initialized BlockManager: BlockManagerId(driver, de-bootcamp.europe-west1-b.c.de-bootcamp-339509.internal, 44699, None)\n"
     ]
    }
   ],
   "source": [
    "sc = SparkContext(conf=conf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "277ed94d",
   "metadata": {},
   "outputs": [],
   "source": [
    "sc._jsc.hadoopConfiguration().set(\"fs.AbstractFileSystem.gs.impl\",  \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS\")\n",
    "sc._jsc.hadoopConfiguration().set(\"fs.gs.impl\", \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem\")\n",
    "sc._jsc.hadoopConfiguration().set(\"fs.gs.auth.service.account.json.keyfile\", \"/home/gerry/.google/credentials/google_credentials.json\")\n",
    "sc._jsc.hadoopConfiguration().set(\"fs.gs.auth.service.account.enable\", \"true\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "806a09fa",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "22/04/02 14:02:02 INFO SharedState: Setting hive.metastore.warehouse.dir ('null') to the value of spark.sql.warehouse.dir ('file:/home/gerry/uk_crime_data/notebooks/spark-warehouse').\n",
      "22/04/02 14:02:02 INFO SharedState: Warehouse path is 'file:/home/gerry/uk_crime_data/notebooks/spark-warehouse'.\n"
     ]
    }
   ],
   "source": [
    "spark = SparkSession.builder \\\n",
    "    .config(conf=sc.getConf()) \\\n",
    "    .getOrCreate()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "fa130136",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'gs://dtc_data_lake_de-bootcamp-339509/metropolitan_2020-01.parquet'"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "bucket_name = \"dtc_data_lake_de-bootcamp-339509\"\n",
    "path = f\"gs://{bucket_name}/metropolitan_2020-01.parquet\"\n",
    "path"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "fa682b0a",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "22/04/02 14:02:09 INFO InMemoryFileIndex: It took 146 ms to list leaf files for 1 paths.\n",
      "22/04/02 14:02:10 INFO SparkContext: Starting job: parquet at NativeMethodAccessorImpl.java:0\n",
      "22/04/02 14:02:10 INFO DAGScheduler: Got job 0 (parquet at NativeMethodAccessorImpl.java:0) with 1 output partitions\n",
      "22/04/02 14:02:10 INFO DAGScheduler: Final stage: ResultStage 0 (parquet at NativeMethodAccessorImpl.java:0)\n",
      "22/04/02 14:02:10 INFO DAGScheduler: Parents of final stage: List()\n",
      "22/04/02 14:02:10 INFO DAGScheduler: Missing parents: List()\n",
      "22/04/02 14:02:10 INFO DAGScheduler: Submitting ResultStage 0 (MapPartitionsRDD[1] at parquet at NativeMethodAccessorImpl.java:0), which has no missing parents\n",
      "22/04/02 14:02:10 INFO MemoryStore: Block broadcast_0 stored as values in memory (estimated size 84.9 KiB, free 434.3 MiB)\n",
      "22/04/02 14:02:10 INFO MemoryStore: Block broadcast_0_piece0 stored as bytes in memory (estimated size 30.3 KiB, free 434.3 MiB)\n",
      "22/04/02 14:02:10 INFO BlockManagerInfo: Added broadcast_0_piece0 in memory on de-bootcamp.europe-west1-b.c.de-bootcamp-339509.internal:44699 (size: 30.3 KiB, free: 434.4 MiB)\n",
      "22/04/02 14:02:10 INFO SparkContext: Created broadcast 0 from broadcast at DAGScheduler.scala:1223\n",
      "22/04/02 14:02:10 INFO DAGScheduler: Submitting 1 missing tasks from ResultStage 0 (MapPartitionsRDD[1] at parquet at NativeMethodAccessorImpl.java:0) (first 15 tasks are for partitions Vector(0))\n",
      "22/04/02 14:02:10 INFO TaskSchedulerImpl: Adding task set 0.0 with 1 tasks\n",
      "22/04/02 14:02:10 INFO TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0, de-bootcamp.europe-west1-b.c.de-bootcamp-339509.internal, executor driver, partition 0, PROCESS_LOCAL, 7514 bytes)\n",
      "22/04/02 14:02:10 INFO Executor: Running task 0.0 in stage 0.0 (TID 0)\n",
      "22/04/02 14:02:11 INFO Executor: Finished task 0.0 in stage 0.0 (TID 0). 2535 bytes result sent to driver\n",
      "22/04/02 14:02:11 INFO TaskSetManager: Finished task 0.0 in stage 0.0 (TID 0) in 1062 ms on de-bootcamp.europe-west1-b.c.de-bootcamp-339509.internal (executor driver) (1/1)\n",
      "22/04/02 14:02:11 INFO DAGScheduler: ResultStage 0 (parquet at NativeMethodAccessorImpl.java:0) finished in 1.409 s\n",
      "22/04/02 14:02:11 INFO TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks have all completed, from pool \n",
      "22/04/02 14:02:11 INFO DAGScheduler: Job 0 is finished. Cancelling potential speculative or zombie tasks for this job\n",
      "22/04/02 14:02:11 INFO TaskSchedulerImpl: Killing all running tasks in stage 0: Stage finished\n",
      "22/04/02 14:02:11 INFO DAGScheduler: Job 0 finished: parquet at NativeMethodAccessorImpl.java:0, took 1.499585 s\n",
      "22/04/02 14:02:13 INFO BlockManagerInfo: Removed broadcast_0_piece0 on de-bootcamp.europe-west1-b.c.de-bootcamp-339509.internal:44699 in memory (size: 30.3 KiB, free: 434.4 MiB)\n"
     ]
    }
   ],
   "source": [
    "df  = spark.read.parquet(path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "e54b7d55",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "22/04/02 13:45:40 INFO FileSourceStrategy: Pushed Filters: \n",
      "22/04/02 13:45:40 INFO FileSourceStrategy: Post-Scan Filters: \n",
      "22/04/02 13:45:40 INFO FileSourceStrategy: Output Data Schema: struct<age_range: string, outcome: string, involved_person: boolean, self_defined_ethnicity: string, gender: string ... 19 more fields>\n",
      "22/04/02 13:45:41 INFO CodeGenerator: Code generated in 401.677735 ms\n",
      "22/04/02 13:45:41 INFO MemoryStore: Block broadcast_1 stored as values in memory (estimated size 180.8 KiB, free 434.2 MiB)\n",
      "22/04/02 13:45:41 INFO MemoryStore: Block broadcast_1_piece0 stored as bytes in memory (estimated size 29.3 KiB, free 434.2 MiB)\n",
      "22/04/02 13:45:41 INFO BlockManagerInfo: Added broadcast_1_piece0 in memory on de-bootcamp.europe-west1-b.c.de-bootcamp-339509.internal:39923 (size: 29.3 KiB, free: 434.4 MiB)\n",
      "22/04/02 13:45:41 INFO SparkContext: Created broadcast 1 from showString at NativeMethodAccessorImpl.java:0\n",
      "22/04/02 13:45:41 INFO FileSourceScanExec: Planning scan with bin packing, max size: 4194304 bytes, open cost is considered as scanning 4194304 bytes.\n",
      "22/04/02 13:45:41 INFO SparkContext: Starting job: showString at NativeMethodAccessorImpl.java:0\n",
      "22/04/02 13:45:41 INFO DAGScheduler: Got job 1 (showString at NativeMethodAccessorImpl.java:0) with 1 output partitions\n",
      "22/04/02 13:45:41 INFO DAGScheduler: Final stage: ResultStage 1 (showString at NativeMethodAccessorImpl.java:0)\n",
      "22/04/02 13:45:41 INFO DAGScheduler: Parents of final stage: List()\n",
      "22/04/02 13:45:41 INFO DAGScheduler: Missing parents: List()\n",
      "22/04/02 13:45:41 INFO DAGScheduler: Submitting ResultStage 1 (MapPartitionsRDD[5] at showString at NativeMethodAccessorImpl.java:0), which has no missing parents\n",
      "22/04/02 13:45:41 INFO MemoryStore: Block broadcast_2 stored as values in memory (estimated size 25.9 KiB, free 434.2 MiB)\n",
      "22/04/02 13:45:41 INFO MemoryStore: Block broadcast_2_piece0 stored as bytes in memory (estimated size 8.2 KiB, free 434.2 MiB)\n",
      "22/04/02 13:45:41 INFO BlockManagerInfo: Added broadcast_2_piece0 in memory on de-bootcamp.europe-west1-b.c.de-bootcamp-339509.internal:39923 (size: 8.2 KiB, free: 434.4 MiB)\n",
      "22/04/02 13:45:41 INFO SparkContext: Created broadcast 2 from broadcast at DAGScheduler.scala:1223\n",
      "22/04/02 13:45:41 INFO DAGScheduler: Submitting 1 missing tasks from ResultStage 1 (MapPartitionsRDD[5] at showString at NativeMethodAccessorImpl.java:0) (first 15 tasks are for partitions Vector(0))\n",
      "22/04/02 13:45:41 INFO TaskSchedulerImpl: Adding task set 1.0 with 1 tasks\n",
      "22/04/02 13:45:41 INFO TaskSetManager: Starting task 0.0 in stage 1.0 (TID 1, de-bootcamp.europe-west1-b.c.de-bootcamp-339509.internal, executor driver, partition 0, PROCESS_LOCAL, 7763 bytes)\n",
      "22/04/02 13:45:41 INFO Executor: Running task 0.0 in stage 1.0 (TID 1)\n",
      "22/04/02 13:45:41 INFO FileScanRDD: Reading File path: gs://dtc_data_lake_de-bootcamp-339509/metropolitan_2020-01.parquet, range: 0-1025904, partition values: [empty row]\n",
      "22/04/02 13:45:42 INFO CodecPool: Got brand-new decompressor [.snappy] + 1) / 1]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------------------+---------------+----------------------+------+--------------------+----------------------------------+-------------------+-----------------------------------+--------------------+--------------------+---------+-------------------------+--------------------+--------------+--------------------+---------+---------+-------------------+---------+--------------------+\n",
      "|age_range|             outcome|involved_person|self_defined_ethnicity|gender|         legislation|outcome_linked_to_object_of_search|           datetime|removal_of_more_than_outer_clothing|      outcome_object|            location|operation|officer_defined_ethnicity|                type|operation_name|    object_of_search| latitude|longitude|           lat_long|street_id|         street_name|\n",
      "+---------+--------------------+---------------+----------------------+------+--------------------+----------------------------------+-------------------+-----------------------------------+--------------------+--------------------+---------+-------------------------+--------------------+--------------+--------------------+---------+---------+-------------------+---------+--------------------+\n",
      "|  over 34|              Arrest|           true|  White - English/W...|  Male|Police and Crimin...|                              null|2020-01-03 18:22:00|                               null|{'id': 'bu-arrest...|                    |    false|                    White|       Person search|          null|        Stolen goods|     null|     null|                   |     null|                    |\n",
      "|    25-34|A no further acti...|           true|  White - English/W...|  Male|Police and Crimin...|                              null|2020-01-05 01:42:00|                               null|{'id': 'bu-no-fur...|{'latitude': '51....|    false|                    White|       Person search|          null|        Stolen goods|51.471396|-0.455787|51.471396,-0.455787|   944283|On or near Parkin...|\n",
      "|    25-34|              Arrest|           true|  Other ethnic grou...|  Male|Police and Crimin...|                              null|2020-01-06 01:40:00|                               null|{'id': 'bu-arrest...|{'latitude': '51....|    false|                    White|       Person search|          null|        Stolen goods| 51.45978| -0.44773|51.459780,-0.447730|   944230|On or near Airpor...|\n",
      "|  over 34|A no further acti...|           true|  Other ethnic grou...|  Male|Firearms Act 1968...|                              null|2020-01-06 01:40:00|                               null|{'id': 'bu-no-fur...|{'latitude': '51....|    false|                    Asian|       Person search|          null|            Firearms|51.470361|-0.458471|51.470361,-0.458471|   944232|On or near Airpor...|\n",
      "|    18-24|              Arrest|           true|  Other ethnic grou...|  Male|Police and Crimin...|                              null|2020-01-06 22:00:00|                               null|{'id': 'bu-arrest...|{'latitude': '51....|    false|                    Black|       Person search|          null|   Offensive weapons|51.471396|-0.455787|51.471396,-0.455787|   944283|On or near Parkin...|\n",
      "|  over 34|A no further acti...|           true|  White - English/W...|  Male|Police and Crimin...|                              null|2020-01-07 11:50:00|                               null|{'id': 'bu-no-fur...|{'latitude': '51....|    false|                    White|       Person search|          null|        Stolen goods|51.471717|-0.490059|51.471717,-0.490059|   944252|On or near Wallis...|\n",
      "|    25-34|A no further acti...|           true|  Other ethnic grou...|  Male|Police and Crimin...|                              null|2020-01-08 12:38:00|                               null|{'id': 'bu-no-fur...|                    |    false|                    White|       Person search|          null|Evidence of offen...|     null|     null|                   |     null|                    |\n",
      "|    18-24|A no further acti...|           true|  White - English/W...|  Male|Police and Crimin...|                              null|2020-01-10 14:57:00|                               null|{'id': 'bu-no-fur...|{'latitude': '51....|    false|                    White|       Person search|          null|Evidence of offen...|51.461945|-0.490972|51.461945,-0.490972|  1160906|On or near Park/O...|\n",
      "|    18-24|A no further acti...|           true|  White - English/W...|  Male|Police and Crimin...|                              null|2020-01-10 14:57:00|                               null|{'id': 'bu-no-fur...|{'latitude': '51....|    false|                    White|       Person search|          null|Evidence of offen...|51.461945|-0.490972|51.461945,-0.490972|  1160906|On or near Park/O...|\n",
      "|    10-17|A no further acti...|           true|  White - English/W...|  Male|Police and Crimin...|                              null|2020-01-10 15:12:00|                               null|{'id': 'bu-no-fur...|{'latitude': '51....|    false|                    White|       Person search|          null|Evidence of offen...|51.461945|-0.490972|51.461945,-0.490972|  1160906|On or near Park/O...|\n",
      "|    10-17|A no further acti...|           true|  White - English/W...|  Male|Police and Crimin...|                              null|2020-01-10 15:12:00|                               null|{'id': 'bu-no-fur...|{'latitude': '51....|    false|                    White|       Person search|          null|Evidence of offen...|51.461945|-0.490972|51.461945,-0.490972|  1160906|On or near Park/O...|\n",
      "|    10-17|A no further acti...|           true|  White - English/W...|  Male|Police and Crimin...|                              null|2020-01-10 15:05:00|                               null|{'id': 'bu-no-fur...|{'latitude': '51....|    false|                    White|       Person search|          null|        Stolen goods|51.461945|-0.490972|51.461945,-0.490972|  1160906|On or near Park/O...|\n",
      "|    10-17|A no further acti...|           true|  White - English/W...|  Male|Police and Crimin...|                              null|2020-01-10 15:05:00|                               null|{'id': 'bu-no-fur...|{'latitude': '51....|    false|                    White|       Person search|          null|        Stolen goods|51.461945|-0.490972|51.461945,-0.490972|  1160906|On or near Park/O...|\n",
      "|    18-24|A no further acti...|           true|  Black/African/Car...|  Male|Misuse of Drugs A...|                              null|2020-01-11 17:30:00|                               null|{'id': 'bu-no-fur...|{'latitude': '51....|    false|                    Black|       Person search|          null|    Controlled drugs|51.501083| 0.022916| 51.501083,0.022916|   954421|On or near Bradfi...|\n",
      "|    18-24|A no further acti...|           true|  Black/African/Car...|  Male|Misuse of Drugs A...|                              null|2020-01-11 17:30:00|                               null|{'id': 'bu-no-fur...|{'latitude': '51....|    false|                    Black|       Person search|          null|    Controlled drugs|51.501083| 0.022916| 51.501083,0.022916|   954421|On or near Bradfi...|\n",
      "|    18-24|A no further acti...|           true|  Asian/Asian Briti...|  Male|Misuse of Drugs A...|                              null|2020-01-11 17:30:00|                               null|{'id': 'bu-no-fur...|{'latitude': '51....|    false|                    Asian|       Person search|          null|    Controlled drugs|51.501083| 0.022916| 51.501083,0.022916|   954421|On or near Bradfi...|\n",
      "|    18-24|A no further acti...|           true|  Asian/Asian Briti...|  Male|Misuse of Drugs A...|                              null|2020-01-11 17:30:00|                               null|{'id': 'bu-no-fur...|{'latitude': '51....|    false|                    Asian|       Person search|          null|    Controlled drugs|51.501083| 0.022916| 51.501083,0.022916|   954421|On or near Bradfi...|\n",
      "|    25-34|              Arrest|           true|  Black/African/Car...|  Male|Misuse of Drugs A...|                              null|2020-01-11 21:30:00|                               null|{'id': 'bu-arrest...|{'latitude': '51....|    false|                    Black|Person and Vehicl...|          null|    Controlled drugs|51.516415|  0.08207| 51.516415,0.082070|   962516|On or near Shoppi...|\n",
      "|    25-34|              Arrest|           true|  Black/African/Car...|  Male|Misuse of Drugs A...|                              null|2020-01-11 21:30:00|                               null|{'id': 'bu-arrest...|{'latitude': '51....|    false|                    Black|Person and Vehicl...|          null|    Controlled drugs|51.516415|  0.08207| 51.516415,0.082070|   962516|On or near Shoppi...|\n",
      "|    18-24|A no further acti...|           true|  White - English/W...|  Male|Misuse of Drugs A...|                              null|2020-01-11 17:35:00|                               null|{'id': 'bu-no-fur...|{'latitude': '51....|    false|                    White|       Person search|          null|    Controlled drugs|51.501083| 0.022916| 51.501083,0.022916|   954421|On or near Bradfi...|\n",
      "+---------+--------------------+---------------+----------------------+------+--------------------+----------------------------------+-------------------+-----------------------------------+--------------------+--------------------+---------+-------------------------+--------------------+--------------+--------------------+---------+---------+-------------------+---------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "22/04/02 13:45:42 INFO Executor: Finished task 0.0 in stage 1.0 (TID 1). 4299 bytes result sent to driver\n",
      "22/04/02 13:45:43 INFO TaskSetManager: Finished task 0.0 in stage 1.0 (TID 1) in 1431 ms on de-bootcamp.europe-west1-b.c.de-bootcamp-339509.internal (executor driver) (1/1)\n",
      "22/04/02 13:45:43 INFO TaskSchedulerImpl: Removed TaskSet 1.0, whose tasks have all completed, from pool \n",
      "22/04/02 13:45:43 INFO DAGScheduler: ResultStage 1 (showString at NativeMethodAccessorImpl.java:0) finished in 1.475 s\n",
      "22/04/02 13:45:43 INFO DAGScheduler: Job 1 is finished. Cancelling potential speculative or zombie tasks for this job\n",
      "22/04/02 13:45:43 INFO TaskSchedulerImpl: Killing all running tasks in stage 1: Stage finished\n",
      "22/04/02 13:45:43 INFO DAGScheduler: Job 1 finished: showString at NativeMethodAccessorImpl.java:0, took 1.483682 s\n",
      "\r",
      "                                                                                \r",
      "22/04/02 13:45:43 INFO CodeGenerator: Code generated in 54.479083 ms\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "5b699c35",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- age_range: string (nullable = true)\n",
      " |-- outcome: string (nullable = true)\n",
      " |-- involved_person: boolean (nullable = true)\n",
      " |-- self_defined_ethnicity: string (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      " |-- legislation: string (nullable = true)\n",
      " |-- outcome_linked_to_object_of_search: string (nullable = true)\n",
      " |-- datetime: timestamp (nullable = true)\n",
      " |-- removal_of_more_than_outer_clothing: string (nullable = true)\n",
      " |-- outcome_object: string (nullable = true)\n",
      " |-- location: string (nullable = true)\n",
      " |-- operation: string (nullable = true)\n",
      " |-- officer_defined_ethnicity: string (nullable = true)\n",
      " |-- type: string (nullable = true)\n",
      " |-- operation_name: string (nullable = true)\n",
      " |-- object_of_search: string (nullable = true)\n",
      " |-- latitude: double (nullable = true)\n",
      " |-- longitude: double (nullable = true)\n",
      " |-- lat_long: string (nullable = true)\n",
      " |-- street_id: long (nullable = true)\n",
      " |-- street_name: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "86e37ac5",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "22/04/02 13:54:29 INFO FileSourceStrategy: Pushed Filters: \n",
      "22/04/02 13:54:29 INFO FileSourceStrategy: Post-Scan Filters: \n",
      "22/04/02 13:54:29 INFO FileSourceStrategy: Output Data Schema: struct<age_range: string, outcome: string, involved_person: boolean, self_defined_ethnicity: string, gender: string ... 19 more fields>\n",
      "22/04/02 13:54:29 INFO ParquetFileFormat: Using default output committer for Parquet: org.apache.parquet.hadoop.ParquetOutputCommitter\n",
      "22/04/02 13:54:29 INFO FileOutputCommitter: File Output Committer Algorithm version is 1\n",
      "22/04/02 13:54:29 INFO FileOutputCommitter: FileOutputCommitter skip cleanup _temporary folders under output directory:false, ignore cleanup failures: false\n",
      "22/04/02 13:54:29 INFO SQLHadoopMapReduceCommitProtocol: Using user defined output committer class org.apache.parquet.hadoop.ParquetOutputCommitter\n",
      "22/04/02 13:54:29 INFO FileOutputCommitter: File Output Committer Algorithm version is 1\n",
      "22/04/02 13:54:29 INFO FileOutputCommitter: FileOutputCommitter skip cleanup _temporary folders under output directory:false, ignore cleanup failures: false\n",
      "22/04/02 13:54:29 INFO SQLHadoopMapReduceCommitProtocol: Using output committer class org.apache.parquet.hadoop.ParquetOutputCommitter\n",
      "22/04/02 13:54:29 INFO CodeGenerator: Code generated in 51.400155 ms\n",
      "22/04/02 13:54:29 INFO MemoryStore: Block broadcast_3 stored as values in memory (estimated size 180.8 KiB, free 434.0 MiB)\n",
      "22/04/02 13:54:29 INFO MemoryStore: Block broadcast_3_piece0 stored as bytes in memory (estimated size 29.3 KiB, free 434.0 MiB)\n",
      "22/04/02 13:54:29 INFO BlockManagerInfo: Added broadcast_3_piece0 in memory on de-bootcamp.europe-west1-b.c.de-bootcamp-339509.internal:39923 (size: 29.3 KiB, free: 434.3 MiB)\n",
      "22/04/02 13:54:29 INFO SparkContext: Created broadcast 3 from parquet at NativeMethodAccessorImpl.java:0\n",
      "22/04/02 13:54:29 INFO FileSourceScanExec: Planning scan with bin packing, max size: 4194304 bytes, open cost is considered as scanning 4194304 bytes.\n",
      "22/04/02 13:54:29 INFO SparkContext: Starting job: parquet at NativeMethodAccessorImpl.java:0\n",
      "22/04/02 13:54:29 INFO DAGScheduler: Got job 2 (parquet at NativeMethodAccessorImpl.java:0) with 1 output partitions\n",
      "22/04/02 13:54:29 INFO DAGScheduler: Final stage: ResultStage 2 (parquet at NativeMethodAccessorImpl.java:0)\n",
      "22/04/02 13:54:29 INFO DAGScheduler: Parents of final stage: List()\n",
      "22/04/02 13:54:29 INFO DAGScheduler: Missing parents: List()\n",
      "22/04/02 13:54:29 INFO DAGScheduler: Submitting ResultStage 2 (MapPartitionsRDD[8] at parquet at NativeMethodAccessorImpl.java:0), which has no missing parents\n",
      "22/04/02 13:54:29 INFO MemoryStore: Block broadcast_4 stored as values in memory (estimated size 191.2 KiB, free 433.8 MiB)\n",
      "22/04/02 13:54:29 INFO MemoryStore: Block broadcast_4_piece0 stored as bytes in memory (estimated size 65.5 KiB, free 433.7 MiB)\n",
      "22/04/02 13:54:29 INFO BlockManagerInfo: Added broadcast_4_piece0 in memory on de-bootcamp.europe-west1-b.c.de-bootcamp-339509.internal:39923 (size: 65.5 KiB, free: 434.3 MiB)\n",
      "22/04/02 13:54:29 INFO SparkContext: Created broadcast 4 from broadcast at DAGScheduler.scala:1223\n",
      "22/04/02 13:54:29 INFO DAGScheduler: Submitting 1 missing tasks from ResultStage 2 (MapPartitionsRDD[8] at parquet at NativeMethodAccessorImpl.java:0) (first 15 tasks are for partitions Vector(0))\n",
      "22/04/02 13:54:29 INFO TaskSchedulerImpl: Adding task set 2.0 with 1 tasks\n",
      "22/04/02 13:54:29 INFO TaskSetManager: Starting task 0.0 in stage 2.0 (TID 2, de-bootcamp.europe-west1-b.c.de-bootcamp-339509.internal, executor driver, partition 0, PROCESS_LOCAL, 7763 bytes)\n",
      "22/04/02 13:54:29 INFO Executor: Running task 0.0 in stage 2.0 (TID 2)\n",
      "22/04/02 13:54:29 INFO FileOutputCommitter: File Output Committer Algorithm version is 1\n",
      "22/04/02 13:54:29 INFO FileOutputCommitter: FileOutputCommitter skip cleanup _temporary folders under output directory:false, ignore cleanup failures: false\n",
      "22/04/02 13:54:29 INFO SQLHadoopMapReduceCommitProtocol: Using user defined output committer class org.apache.parquet.hadoop.ParquetOutputCommitter\n",
      "22/04/02 13:54:29 INFO FileOutputCommitter: File Output Committer Algorithm version is 1\n",
      "22/04/02 13:54:29 INFO FileOutputCommitter: FileOutputCommitter skip cleanup _temporary folders under output directory:false, ignore cleanup failures: false\n",
      "22/04/02 13:54:29 INFO SQLHadoopMapReduceCommitProtocol: Using output committer class org.apache.parquet.hadoop.ParquetOutputCommitter\n",
      "22/04/02 13:54:29 INFO CodecConfig: Compression: SNAPPY\n",
      "22/04/02 13:54:29 INFO CodecConfig: Compression: SNAPPY\n",
      "22/04/02 13:54:29 INFO ParquetOutputFormat: Parquet block size to 134217728\n",
      "22/04/02 13:54:29 INFO ParquetOutputFormat: Parquet page size to 1048576\n",
      "22/04/02 13:54:29 INFO ParquetOutputFormat: Parquet dictionary page size to 1048576\n",
      "22/04/02 13:54:29 INFO ParquetOutputFormat: Dictionary is on\n",
      "22/04/02 13:54:29 INFO ParquetOutputFormat: Validation is off\n",
      "22/04/02 13:54:29 INFO ParquetOutputFormat: Writer version is: PARQUET_1_0\n",
      "22/04/02 13:54:29 INFO ParquetOutputFormat: Maximum row group padding size is 8388608 bytes\n",
      "22/04/02 13:54:29 INFO ParquetOutputFormat: Page size checking is: estimated\n",
      "22/04/02 13:54:29 INFO ParquetOutputFormat: Min row count for page size check is: 100\n",
      "22/04/02 13:54:29 INFO ParquetOutputFormat: Max row count for page size check is: 10000\n",
      "22/04/02 13:54:29 INFO ParquetWriteSupport: Initialized Parquet WriteSupport with Catalyst schema:\n",
      "{\n",
      "  \"type\" : \"struct\",\n",
      "  \"fields\" : [ {\n",
      "    \"name\" : \"age_range\",\n",
      "    \"type\" : \"string\",\n",
      "    \"nullable\" : true,\n",
      "    \"metadata\" : { }\n",
      "  }, {\n",
      "    \"name\" : \"outcome\",\n",
      "    \"type\" : \"string\",\n",
      "    \"nullable\" : true,\n",
      "    \"metadata\" : { }\n",
      "  }, {\n",
      "    \"name\" : \"involved_person\",\n",
      "    \"type\" : \"boolean\",\n",
      "    \"nullable\" : true,\n",
      "    \"metadata\" : { }\n",
      "  }, {\n",
      "    \"name\" : \"self_defined_ethnicity\",\n",
      "    \"type\" : \"string\",\n",
      "    \"nullable\" : true,\n",
      "    \"metadata\" : { }\n",
      "  }, {\n",
      "    \"name\" : \"gender\",\n",
      "    \"type\" : \"string\",\n",
      "    \"nullable\" : true,\n",
      "    \"metadata\" : { }\n",
      "  }, {\n",
      "    \"name\" : \"legislation\",\n",
      "    \"type\" : \"string\",\n",
      "    \"nullable\" : true,\n",
      "    \"metadata\" : { }\n",
      "  }, {\n",
      "    \"name\" : \"outcome_linked_to_object_of_search\",\n",
      "    \"type\" : \"string\",\n",
      "    \"nullable\" : true,\n",
      "    \"metadata\" : { }\n",
      "  }, {\n",
      "    \"name\" : \"datetime\",\n",
      "    \"type\" : \"timestamp\",\n",
      "    \"nullable\" : true,\n",
      "    \"metadata\" : { }\n",
      "  }, {\n",
      "    \"name\" : \"removal_of_more_than_outer_clothing\",\n",
      "    \"type\" : \"string\",\n",
      "    \"nullable\" : true,\n",
      "    \"metadata\" : { }\n",
      "  }, {\n",
      "    \"name\" : \"outcome_object\",\n",
      "    \"type\" : \"string\",\n",
      "    \"nullable\" : true,\n",
      "    \"metadata\" : { }\n",
      "  }, {\n",
      "    \"name\" : \"location\",\n",
      "    \"type\" : \"string\",\n",
      "    \"nullable\" : true,\n",
      "    \"metadata\" : { }\n",
      "  }, {\n",
      "    \"name\" : \"operation\",\n",
      "    \"type\" : \"string\",\n",
      "    \"nullable\" : true,\n",
      "    \"metadata\" : { }\n",
      "  }, {\n",
      "    \"name\" : \"officer_defined_ethnicity\",\n",
      "    \"type\" : \"string\",\n",
      "    \"nullable\" : true,\n",
      "    \"metadata\" : { }\n",
      "  }, {\n",
      "    \"name\" : \"type\",\n",
      "    \"type\" : \"string\",\n",
      "    \"nullable\" : true,\n",
      "    \"metadata\" : { }\n",
      "  }, {\n",
      "    \"name\" : \"operation_name\",\n",
      "    \"type\" : \"string\",\n",
      "    \"nullable\" : true,\n",
      "    \"metadata\" : { }\n",
      "  }, {\n",
      "    \"name\" : \"object_of_search\",\n",
      "    \"type\" : \"string\",\n",
      "    \"nullable\" : true,\n",
      "    \"metadata\" : { }\n",
      "  }, {\n",
      "    \"name\" : \"latitude\",\n",
      "    \"type\" : \"double\",\n",
      "    \"nullable\" : true,\n",
      "    \"metadata\" : { }\n",
      "  }, {\n",
      "    \"name\" : \"longitude\",\n",
      "    \"type\" : \"double\",\n",
      "    \"nullable\" : true,\n",
      "    \"metadata\" : { }\n",
      "  }, {\n",
      "    \"name\" : \"lat_long\",\n",
      "    \"type\" : \"string\",\n",
      "    \"nullable\" : true,\n",
      "    \"metadata\" : { }\n",
      "  }, {\n",
      "    \"name\" : \"street_id\",\n",
      "    \"type\" : \"long\",\n",
      "    \"nullable\" : true,\n",
      "    \"metadata\" : { }\n",
      "  }, {\n",
      "    \"name\" : \"street_name\",\n",
      "    \"type\" : \"string\",\n",
      "    \"nullable\" : true,\n",
      "    \"metadata\" : { }\n",
      "  } ]\n",
      "}\n",
      "and corresponding Parquet message type:\n",
      "message spark_schema {\n",
      "  optional binary age_range (UTF8);\n",
      "  optional binary outcome (UTF8);\n",
      "  optional boolean involved_person;\n",
      "  optional binary self_defined_ethnicity (UTF8);\n",
      "  optional binary gender (UTF8);\n",
      "  optional binary legislation (UTF8);\n",
      "  optional binary outcome_linked_to_object_of_search (UTF8);\n",
      "  optional int96 datetime;\n",
      "  optional binary removal_of_more_than_outer_clothing (UTF8);\n",
      "  optional binary outcome_object (UTF8);\n",
      "  optional binary location (UTF8);\n",
      "  optional binary operation (UTF8);\n",
      "  optional binary officer_defined_ethnicity (UTF8);\n",
      "  optional binary type (UTF8);\n",
      "  optional binary operation_name (UTF8);\n",
      "  optional binary object_of_search (UTF8);\n",
      "  optional double latitude;\n",
      "  optional double longitude;\n",
      "  optional binary lat_long (UTF8);\n",
      "  optional int64 street_id;\n",
      "  optional binary street_name (UTF8);\n",
      "}\n",
      "\n",
      "       \n",
      "22/04/02 13:54:29 INFO CodecPool: Got brand-new compressor [.snappy]\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "22/04/02 13:54:29 INFO FileScanRDD: Reading File path: gs://dtc_data_lake_de-bootcamp-339509/metropolitan_2020-01.parquet, range: 0-1025904, partition values: [empty row]\n",
      "22/04/02 13:54:31 INFO BlockManagerInfo: Removed broadcast_2_piece0 on de-bootcamp.europe-west1-b.c.de-bootcamp-339509.internal:39923 in memory (size: 8.2 KiB, free: 434.3 MiB)\n",
      "22/04/02 13:54:31 INFO BlockManagerInfo: Removed broadcast_1_piece0 on de-bootcamp.europe-west1-b.c.de-bootcamp-339509.internal:39923 in memory (size: 29.3 KiB, free: 434.3 MiB)\n",
      "22/04/02 13:54:31 INFO InternalParquetRecordWriter: Flushing mem columnStore to file. allocated memory: 3174311\n",
      "22/04/02 13:54:31 INFO FileOutputCommitter: Saved output of task 'attempt_20220402135429444884042504427888_0002_m_000000_2' to file:/home/gerry/uk_crime_data/notebooks/new_data2/_temporary/0/task_20220402135429444884042504427888_0002_m_000000\n",
      "22/04/02 13:54:31 INFO SparkHadoopMapRedUtil: attempt_20220402135429444884042504427888_0002_m_000000_2: Committed\n",
      "22/04/02 13:54:31 INFO Executor: Finished task 0.0 in stage 2.0 (TID 2). 2570 bytes result sent to driver\n",
      "22/04/02 13:54:31 INFO TaskSetManager: Finished task 0.0 in stage 2.0 (TID 2) in 2065 ms on de-bootcamp.europe-west1-b.c.de-bootcamp-339509.internal (executor driver) (1/1)\n",
      "22/04/02 13:54:31 INFO TaskSchedulerImpl: Removed TaskSet 2.0, whose tasks have all completed, from pool \n",
      "22/04/02 13:54:31 INFO DAGScheduler: ResultStage 2 (parquet at NativeMethodAccessorImpl.java:0) finished in 2.106 s\n",
      "22/04/02 13:54:31 INFO DAGScheduler: Job 2 is finished. Cancelling potential speculative or zombie tasks for this job\n",
      "22/04/02 13:54:31 INFO TaskSchedulerImpl: Killing all running tasks in stage 2: Stage finished\n",
      "22/04/02 13:54:31 INFO DAGScheduler: Job 2 finished: parquet at NativeMethodAccessorImpl.java:0, took 2.115468 s\n",
      "22/04/02 13:54:31 INFO FileFormatWriter: Write Job 8735689e-3ea5-46d4-a520-722b1631d762 committed.\n",
      "22/04/02 13:54:31 INFO FileFormatWriter: Finished processing stats for write job 8735689e-3ea5-46d4-a520-722b1631d762.\n"
     ]
    }
   ],
   "source": [
    "df = df.write.parquet('new_data2/')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e83983ef",
   "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.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}