escaping_the_fog.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "d0cd4255",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "3.10.9 (main, Mar  1 2023, 18:23:06) [GCC 11.2.0]\n"
     ]
    }
   ],
   "source": [
    "import timeit\n",
    "start = timeit.default_timer()\n",
    "\n",
    "import os\n",
    "os.environ['PYARROW_IGNORE_TIMEZONE'] = '1'\n",
    "import sys\n",
    "print(sys.version)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "d9db932e",
   "metadata": {},
   "outputs": [],
   "source": [
    "import folium\n",
    "from folium.plugins import MarkerCluster\n",
    "import pyarrow as pa\n",
    "import geohash_tools as gh\n",
    "import pyspark as ps"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "b554a101",
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- 1_time: long (nullable = true)\n",
      " |-- 2_lat: double (nullable = true)\n",
      " |-- 3_lon: double (nullable = true)\n",
      " |-- albedo_surface: double (nullable = true)\n",
      " |-- precipitable_water_entire_atmosphere_single_layer: double (nullable = true)\n",
      " |-- pressure_maximum_wind: double (nullable = true)\n",
      " |-- pressure_surface: double (nullable = true)\n",
      " |-- pressure_tropopause: double (nullable = true)\n",
      " |-- relative_humidity_zerodegc_isotherm: double (nullable = true)\n",
      " |-- snow_depth_surface: double (nullable = true)\n",
      " |-- temperature_surface: double (nullable = true)\n",
      " |-- temperature_tropopause: double (nullable = true)\n",
      " |-- total_cloud_cover_entire_atmosphere_single_layer: double (nullable = true)\n",
      " |-- total_precipitation_surface_3_hour_accumulation: double (nullable = true)\n",
      " |-- vegetation_surface: double (nullable = true)\n",
      " |-- visibility_surface: double (nullable = true)\n",
      " |-- wilting_point_surface: double (nullable = true)\n",
      " |-- wind_speed_gust_surface: double (nullable = true)\n",
      " |-- _c18: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sp = spark.read.load('hdfs://orion11:14001/nam/2019/01/05/*',\n",
    "                     format='csv',\n",
    "                     sep='\\t',\n",
    "                     inferSchema='true',\n",
    "                     header='true')\n",
    "\n",
    "schema = sp.schema\n",
    "\n",
    "sp = spark.read.schema(schema).load('hdfs://orion11:14001/nam/201*/*/*/*',\n",
    "                     format='csv',\n",
    "                     sep='\\t',\n",
    "                     header='true')\n",
    "sp.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "3cd36206",
   "metadata": {},
   "outputs": [],
   "source": [
    "sp = sp.select(['2_lat','3_lon','visibility_surface'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "80d6d0f0",
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 23:==================================================>   (115 + 8) / 123]\r"
     ]
    }
   ],
   "source": [
    "fog_sanctuaries = sp.rdd \\\n",
    "    .map(lambda x: (gh.encode(x[0],x[1]), (x[2], 1))) \\\n",
    "    .reduceByKey(lambda a,b: (a[0]+b[0], a[1]+b[1])) \\\n",
    "    .map(lambda x: (x[0], x[1][0]/x[1][1])) \\\n",
    "    .sortBy(lambda x: x[1], ascending=False) \\\n",
    "    .take(40)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "afd6ed99",
   "metadata": {},
   "outputs": [],
   "source": [
    "m = folium.Map(location=[41,-98.58], zoom_start=3)\n",
    "mc = MarkerCluster().add_to(m)\n",
    "\n",
    "for fs in fog_sanctuaries:\n",
    "    folium.Marker(location=list(gh.decode(fs[0])), popup=str(fs[1])).add_to(mc)\n",
    "\n",
    "m"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d5cf2072",
   "metadata": {},
   "outputs": [],
   "source": [
    "print(timeit.default_timer()-start)"
   ]
  }
 ],
 "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.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}