spark/notebooks/geolake-scala-demo.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a07574a6",
   "metadata": {},
   "outputs": [],
   "source": [
    "import org.wololo.geojson.GeoJSONFactory\n",
    "import org.wololo.jts2geojson.GeoJSONReader\n",
    "import org.wololo.geojson.Feature\n",
    "import org.apache.spark.sql.DataFrame\n",
    "\n",
    "def readLineOfGeojson(filePath: String): DataFrame = {\n",
    "    val colnames = Seq(\"TRIP_ID\", \"CALL_TYPE\", \"ORIGIN_STAND\", \"TAXI_ID\", \"TIMESTAMP\", \"DAY_TYPE\", \"MISSING_DATA\",  \"geometry\")\n",
    "    sc.textFile(filePath).map(line => {\n",
    "        val feature = GeoJSONFactory.create(line).asInstanceOf[Feature]\n",
    "        val reader = new GeoJSONReader\n",
    "        (\n",
    "            feature.getProperties.get(\"TRIP_ID\").asInstanceOf[Long]\n",
    "            ,feature.getProperties.get(\"CALL_TYPE\").asInstanceOf[String]\n",
    "            ,feature.getProperties.get(\"ORIGIN_STAND\").asInstanceOf[String]\n",
    "            ,feature.getProperties.get(\"TAXI_ID\").asInstanceOf[Integer]\n",
    "            ,feature.getProperties.get(\"TIMESTAMP\").asInstanceOf[String]\n",
    "            ,feature.getProperties.get(\"DAY_TYPE\").asInstanceOf[String]\n",
    "            ,feature.getProperties.get(\"MISSING_DATA\").asInstanceOf[Boolean]\n",
    "            ,reader.read(feature.getGeometry)\n",
    "        )\n",
    "    }).toDF(colnames: _*)\n",
    "}\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "dc7f1482",
   "metadata": {},
   "outputs": [],
   "source": [
    "val testFile = \"/home/iceberg/data/test_portotaxi.geojson\"\n",
    "val df_test = readLineOfGeojson(testFile)\n",
    "df_test.createOrReplaceTempView(\"test_portotaxi\")\n",
    "df_test.show"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6e85caf5",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_test.printSchema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5fdebf66",
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.sql(\"SELECT * FROM test_portotaxi\").count()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "56f3b175",
   "metadata": {},
   "source": [
    "## 1. Create a iceberg table with geometry type\n",
    "\n",
    "\n",
    "The `write.parquet.geometry.encoding` has 3 possible values:\n",
    "\n",
    "- `nested-list`: the most efficient (usually have small file size, faster reading and writing)\n",
    "\n",
    "- `wkb-bbox`\n",
    "\n",
    "- `wkb`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "14e0f5e8",
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.sql(\"DROP TABLE IF EXISTS demo.db.test_portotaxi\")\n",
    "\n",
    "\n",
    "spark.sql(\"\"\"\n",
    "CREATE TABLE IF NOT EXISTS demo.db.test_portotaxi \n",
    "(\n",
    "  TRIP_ID LONG,\n",
    "  CALL_TYPE STRING,\n",
    "  ORIGIN_STAND STRING,\n",
    "  TAXI_ID INTEGER,\n",
    "  TIMESTAMP STRING,\n",
    "  DAY_TYPE STRING,\n",
    "  MISSING_DATA BOOLEAN,\n",
    "  geometry GEOMETRY\n",
    ")\n",
    "USING iceberg\n",
    "TBLPROPERTIES ('write.parquet.geometry.encoding' = 'nested-list')\n",
    "\"\"\")\n",
    "          \n",
    "spark.sql(\"INSERT INTO demo.db.test_portotaxi SELECT * FROM test_portotaxi\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5e2d3b63",
   "metadata": {},
   "source": [
    "## 2. Table Summary"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "89ce09e3",
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.sql(\"\"\"SELECT\n",
    "    summary['total-records'] as total_records,\n",
    "    summary['total-files-size'] / 1024 / 1024 as file_size_in_mb \n",
    "    FROM demo.db.test_portotaxi.snapshots\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9fc5c202",
   "metadata": {},
   "source": [
    "## 3. Run Spatial Query"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "02ef94e1",
   "metadata": {},
   "outputs": [],
   "source": [
    "val bbox = \"POLYGON ((-8.6079 41.1489, -8.6089 41.1472, -8.6066 41.1470, -8.6061 41.1483, -8.6079 41.1489))\"\n",
    "spark.sql(s\"\"\"\n",
    "SELECT count(*)\n",
    "FROM demo.db.test_portotaxi\n",
    "WHERE ST_Within(geometry, IcebergSTGeomFromText('${bbox}'))\n",
    "\"\"\").show()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "spylon-kernel",
   "language": "scala",
   "name": "spylon-kernel"
  },
  "language_info": {
   "codemirror_mode": "text/x-scala",
   "file_extension": ".scala",
   "help_links": [
    {
     "text": "MetaKernel Magics",
     "url": "https://metakernel.readthedocs.io/en/latest/source/README.html"
    }
   ],
   "mimetype": "text/x-scala",
   "name": "scala",
   "pygments_lexer": "scala",
   "version": "0.4.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}