notebooks/02-ingestion/06-partitioning-data.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "0cb3b009-ebde-4d56-9d59-a028d66d8309",
   "metadata": {},
   "source": [
    "# Optimizing performance using PARTITIONED BY and CLUSTERED BY\n",
    "<!--\n",
    "  ~ Licensed to the Apache Software Foundation (ASF) under one\n",
    "  ~ or more contributor license agreements.  See the NOTICE file\n",
    "  ~ distributed with this work for additional information\n",
    "  ~ regarding copyright ownership.  The ASF licenses this file\n",
    "  ~ to you under the Apache License, Version 2.0 (the\n",
    "  ~ \"License\"); you may not use this file except in compliance\n",
    "  ~ with the License.  You may obtain a copy of the License at\n",
    "  ~\n",
    "  ~   http://www.apache.org/licenses/LICENSE-2.0\n",
    "  ~\n",
    "  ~ Unless required by applicable law or agreed to in writing,\n",
    "  ~ software distributed under the License is distributed on an\n",
    "  ~ \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n",
    "  ~ KIND, either express or implied.  See the License for the\n",
    "  ~ specific language governing permissions and limitations\n",
    "  ~ under the License.\n",
    "  -->\n",
    "\n",
    "This notebook is a review of Apache Druid's data organization strategy and how it can be used to improve system health and query performance.\n",
    "\n",
    "[Apache Druid stores data in Segment files](https://druid.apache.org/docs/latest/design/segments). Druid Segments have a columnar structure designed for high performance analytic queries. Segments are organized into consistent time intervals called the `Segment Granularity` of the table and typically expressed as HOUR, DAY, MONTH, or YEAR. In this notebook you will review how this setting affects the number of segments created during ingestion and the size of those segments. Druid works best when segments are balanced in size with a general rule of thumb that they should cover approximately 5 million rows each. \n",
    "\n",
    "[Pruning](https://druid.apache.org/docs/latest/design/architecture#query-processing) is the process of reducing the segment files that need to be inspected in order to resolve a query. To process a query in Druid, all the time chunks that overlap the query's filter condition on the __time column will be inspected. \n",
    "\n",
    "Within a time chunk, [segment files can be organized based on another set of columns](https://druid.apache.org/docs/latest/ingestion/partitioning). In SQL Based ingestion it's specified using a [CLUSTERED BY clause](https://druid.apache.org/docs/latest/multi-stage-query/reference#clustered-by). When you use this feature, Druid will organize the segments within a time chunk such that each one covers a range of the values from the clustering columns. At query time the Broker can do additional pruning of the segments within a time chunk if the user specifies a filter condition on the clustering columns. "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bbdbf6ad-ca7b-40f5-8ca3-1070f4a3ee42",
   "metadata": {},
   "source": [
    "## Prerequisites\n",
    "\n",
    "This tutorial works was tested with Druid 27.0.0.\n",
    "\n",
    "#### Run with Docker\n",
    "\n",
    "<!-- Profiles are:\n",
    "`druid-jupyter` - just Jupyter and Druid\n",
    "`all-services` - includes Jupyter, Druid, and Kafka\n",
    " -->\n",
    "\n",
    "Launch this tutorial and all prerequisites using the `druid-jupyter` profile of the Docker Compose file for Jupyter-based Druid tutorials. For more information, see [the project on github](https://github.com/implydata/learn-druid).\n",
    "   "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5007a243-b81a-4601-8f57-5b14940abbff",
   "metadata": {},
   "source": [
    "## Initialization\n",
    "\n",
    "The following cells set up the notebook and learning environment ready for use.\n",
    "\n",
    "### Set up and connect to the learning environment\n",
    "\n",
    "Run the next cell to set up the Druid Python client's connection to Apache Druid.\n",
    "\n",
    "If successful, the Druid version number will be shown in the output."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c1ec783b-df3f-4168-9be2-cdc6ad3e33c2",
   "metadata": {},
   "outputs": [],
   "source": [
    "import druidapi\n",
    "import os\n",
    "from datetime import datetime\n",
    "from statistics import mean \n",
    "\n",
    "if 'DRUID_HOST' not in os.environ.keys():\n",
    "    druid_host=f\"http://localhost:8888\"\n",
    "else:\n",
    "    druid_host=f\"http://{os.environ['DRUID_HOST']}:8888\"\n",
    "    \n",
    "print(f\"Opening a connection to {druid_host}.\")\n",
    "druid = druidapi.jupyter_client(druid_host)\n",
    "\n",
    "display = druid.display\n",
    "sql_client = druid.sql\n",
    "status_client = druid.status\n",
    "\n",
    "status_client.version"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2efdbee0-62da-4fd3-84e1-f66b8c0150b3",
   "metadata": {},
   "source": [
    "<!-- Include these cells if your notebook uses Kafka. -->\n",
    "\n",
    "Run the next cell to set up the connection to Apache Kafka."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "472589e4-1026-4b3b-bb79-eedabb2b44c4",
   "metadata": {},
   "source": [
    "## Time partitioning\n",
    "\n",
    "Normally, ideal segments contain around 5 million rows, but given that this is all running on a laptop and for fast demonstration purposes, we'll set our \"ideal\" to only 50000 rows.\n",
    "\n",
    "\n",
    "### Use PARTITIONED BY to adjust the coverage of segments\n",
    "\n",
    "#### Too granular\n",
    "Using a `Segment Granularity` that is too small will render too many segments. The following batch ingestion demonstrates this. It will take about 1 minute to complete:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f52a94fb-d2e4-403f-ab10-84d3af7bf2c8",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql='''\n",
    "REPLACE INTO \"flights-hour\" OVERWRITE ALL\n",
    "WITH \"ext\" AS (SELECT *\n",
    "FROM TABLE(\n",
    "  EXTERN(\n",
    "    '{\"type\":\"http\",\"uris\":[\"https://static.imply.io/example-data/flight_on_time/flights/On_Time_Reporting_Carrier_On_Time_Performance_(1987_present)_2005_11.csv.zip\"]}',\n",
    "    '{\"type\":\"csv\",\"findColumnsFromHeader\":true}'\n",
    "  )\n",
    ") EXTEND (\"depaturetime\" VARCHAR, \"arrivalime\" VARCHAR, \"Year\" BIGINT, \"Quarter\" BIGINT, \"Month\" BIGINT, \"DayofMonth\" BIGINT, \"DayOfWeek\" BIGINT, \"FlightDate\" VARCHAR, \"Reporting_Airline\" VARCHAR, \"DOT_ID_Reporting_Airline\" BIGINT, \"IATA_CODE_Reporting_Airline\" VARCHAR, \"Tail_Number\" VARCHAR, \"Flight_Number_Reporting_Airline\" BIGINT, \"OriginAirportID\" BIGINT, \"OriginAirportSeqID\" BIGINT, \"OriginCityMarketID\" BIGINT, \"Origin\" VARCHAR, \"OriginCityName\" VARCHAR, \"OriginState\" VARCHAR, \"OriginStateFips\" BIGINT, \"OriginStateName\" VARCHAR, \"OriginWac\" BIGINT, \"DestAirportID\" BIGINT, \"DestAirportSeqID\" BIGINT, \"DestCityMarketID\" BIGINT, \"Dest\" VARCHAR, \"DestCityName\" VARCHAR, \"DestState\" VARCHAR, \"DestStateFips\" BIGINT, \"DestStateName\" VARCHAR, \"DestWac\" BIGINT, \"CRSDepTime\" BIGINT, \"DepTime\" BIGINT, \"DepDelay\" BIGINT, \"DepDelayMinutes\" BIGINT, \"DepDel15\" BIGINT, \"DepartureDelayGroups\" BIGINT, \"DepTimeBlk\" VARCHAR, \"TaxiOut\" BIGINT, \"WheelsOff\" BIGINT, \"WheelsOn\" BIGINT, \"TaxiIn\" BIGINT, \"CRSArrTime\" BIGINT, \"ArrTime\" BIGINT, \"ArrDelay\" BIGINT, \"ArrDelayMinutes\" BIGINT, \"ArrDel15\" BIGINT, \"ArrivalDelayGroups\" BIGINT, \"ArrTimeBlk\" VARCHAR, \"Cancelled\" BIGINT, \"CancellationCode\" VARCHAR, \"Diverted\" BIGINT, \"CRSElapsedTime\" BIGINT, \"ActualElapsedTime\" BIGINT, \"AirTime\" BIGINT, \"Flights\" BIGINT, \"Distance\" BIGINT, \"DistanceGroup\" BIGINT, \"CarrierDelay\" BIGINT, \"WeatherDelay\" BIGINT, \"NASDelay\" BIGINT, \"SecurityDelay\" BIGINT, \"LateAircraftDelay\" BIGINT, \"FirstDepTime\" VARCHAR, \"TotalAddGTime\" VARCHAR, \"LongestAddGTime\" VARCHAR, \"DivAirportLandings\" VARCHAR, \"DivReachedDest\" VARCHAR, \"DivActualElapsedTime\" VARCHAR, \"DivArrDelay\" VARCHAR, \"DivDistance\" VARCHAR, \"Div1Airport\" VARCHAR, \"Div1AirportID\" VARCHAR, \"Div1AirportSeqID\" VARCHAR, \"Div1WheelsOn\" VARCHAR, \"Div1TotalGTime\" VARCHAR, \"Div1LongestGTime\" VARCHAR, \"Div1WheelsOff\" VARCHAR, \"Div1TailNum\" VARCHAR, \"Div2Airport\" VARCHAR, \"Div2AirportID\" VARCHAR, \"Div2AirportSeqID\" VARCHAR, \"Div2WheelsOn\" VARCHAR, \"Div2TotalGTime\" VARCHAR, \"Div2LongestGTime\" VARCHAR, \"Div2WheelsOff\" VARCHAR, \"Div2TailNum\" VARCHAR, \"Div3Airport\" VARCHAR, \"Div3AirportID\" VARCHAR, \"Div3AirportSeqID\" VARCHAR, \"Div3WheelsOn\" VARCHAR, \"Div3TotalGTime\" VARCHAR, \"Div3LongestGTime\" VARCHAR, \"Div3WheelsOff\" VARCHAR, \"Div3TailNum\" VARCHAR, \"Div4Airport\" VARCHAR, \"Div4AirportID\" VARCHAR, \"Div4AirportSeqID\" VARCHAR, \"Div4WheelsOn\" VARCHAR, \"Div4TotalGTime\" VARCHAR, \"Div4LongestGTime\" VARCHAR, \"Div4WheelsOff\" VARCHAR, \"Div4TailNum\" VARCHAR, \"Div5Airport\" VARCHAR, \"Div5AirportID\" VARCHAR, \"Div5AirportSeqID\" VARCHAR, \"Div5WheelsOn\" VARCHAR, \"Div5TotalGTime\" VARCHAR, \"Div5LongestGTime\" VARCHAR, \"Div5WheelsOff\" VARCHAR, \"Div5TailNum\" VARCHAR, \"Unnamed: 109\" VARCHAR))\n",
    "SELECT\n",
    "  TIME_PARSE(\"depaturetime\") AS \"__time\",\n",
    "  \"arrivalime\", \"Year\", \"Quarter\", \"Month\", \"DayofMonth\", \"DayOfWeek\", \"FlightDate\", \"Reporting_Airline\", \"DOT_ID_Reporting_Airline\",\n",
    "  \"IATA_CODE_Reporting_Airline\", \"Tail_Number\", \"Flight_Number_Reporting_Airline\", \n",
    "  \"OriginAirportID\", \"OriginAirportSeqID\", \"OriginCityMarketID\", \"Origin\", \"OriginCityName\", \"OriginState\", \"OriginStateFips\", \"OriginStateName\", \"OriginWac\",  \n",
    "  \"DestAirportID\", \"DestAirportSeqID\", \"DestCityMarketID\", \"Dest\", \"DestCityName\", \"DestState\", \"DestStateFips\", \"DestStateName\", \"DestWac\", \n",
    "  \"CRSDepTime\", \"DepTime\", \"DepDelay\", \"DepDelayMinutes\", \"DepDel15\", \"DepartureDelayGroups\", \"DepTimeBlk\",\n",
    "  \"TaxiOut\", \"WheelsOff\", \"WheelsOn\", \"TaxiIn\", \"CRSArrTime\", \"ArrTime\", \"ArrDelay\", \"ArrDelayMinutes\", \"ArrDel15\", \"ArrivalDelayGroups\", \"ArrTimeBlk\",\n",
    "  \"Cancelled\", \"CancellationCode\", \"Diverted\", \"CRSElapsedTime\", \"ActualElapsedTime\", \"AirTime\", \"Flights\", \"Distance\", \"DistanceGroup\", \"CarrierDelay\",\n",
    "  \"WeatherDelay\", \"NASDelay\", \"SecurityDelay\", \"LateAircraftDelay\", \"FirstDepTime\", \"TotalAddGTime\", \"LongestAddGTime\", \"DivAirportLandings\", \"DivReachedDest\",\n",
    "  \"DivActualElapsedTime\",  \"DivArrDelay\",  \"DivDistance\",  \"Div1Airport\",  \"Div1AirportID\",  \"Div1AirportSeqID\",  \"Div1WheelsOn\",  \"Div1TotalGTime\",\n",
    "  \"Div1LongestGTime\",  \"Div1WheelsOff\",  \"Div1TailNum\",  \"Div2Airport\",  \"Div2AirportID\",  \"Div2AirportSeqID\",  \"Div2WheelsOn\",  \"Div2TotalGTime\",\n",
    "  \"Div2LongestGTime\",  \"Div2WheelsOff\",  \"Div2TailNum\",  \"Div3Airport\",  \"Div3AirportID\",  \"Div3AirportSeqID\",  \"Div3WheelsOn\",  \"Div3TotalGTime\",\n",
    "  \"Div3LongestGTime\",  \"Div3WheelsOff\",  \"Div3TailNum\",  \"Div4Airport\",  \"Div4AirportID\",  \"Div4AirportSeqID\",  \"Div4WheelsOn\",  \"Div4TotalGTime\",\n",
    "  \"Div4LongestGTime\",  \"Div4WheelsOff\",  \"Div4TailNum\",  \"Div5Airport\",  \"Div5AirportID\",  \"Div5AirportSeqID\",  \"Div5WheelsOn\",  \"Div5TotalGTime\",\n",
    "  \"Div5LongestGTime\",  \"Div5WheelsOff\",  \"Div5TailNum\"\n",
    "FROM \"ext\"\n",
    "PARTITIONED BY HOUR'''\n",
    "\n",
    "display.run_task(sql)\n",
    "sql_client.wait_until_ready('flights-hour')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "35a923ea-cc64-4b0f-83ce-b5b68722449b",
   "metadata": {},
   "source": [
    "You can see the segments that were created on [the Apache Druid console in the Segments view](http://localhost:8888/unified-console.html#segments/datasource~flights-hour).\n",
    "\n",
    "Here's what it should look like, notice that each segment corresponds to a single time interval of one hour and the number of rows in the segments are small and highly variable:\n",
    "![](assets/segments-hourly.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b4dae0a6-2ae4-474a-9c8d-afc85ad3d909",
   "metadata": {},
   "source": [
    "#### Too coarse\n",
    "In this example you overcorrect by using a granularity of YEAR:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8426f6ec-84ee-4de0-92ac-d90c2df656ad",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql='''\n",
    "REPLACE INTO \"flights-year\" OVERWRITE ALL\n",
    "WITH \"ext\" AS (SELECT *\n",
    "FROM TABLE(\n",
    "  EXTERN(\n",
    "    '{\"type\":\"http\",\"uris\":[\"https://static.imply.io/example-data/flight_on_time/flights/On_Time_Reporting_Carrier_On_Time_Performance_(1987_present)_2005_11.csv.zip\"]}',\n",
    "    '{\"type\":\"csv\",\"findColumnsFromHeader\":true}'\n",
    "  )\n",
    ") EXTEND (\"depaturetime\" VARCHAR, \"arrivalime\" VARCHAR, \"Year\" BIGINT, \"Quarter\" BIGINT, \"Month\" BIGINT, \"DayofMonth\" BIGINT, \"DayOfWeek\" BIGINT, \"FlightDate\" VARCHAR, \"Reporting_Airline\" VARCHAR, \"DOT_ID_Reporting_Airline\" BIGINT, \"IATA_CODE_Reporting_Airline\" VARCHAR, \"Tail_Number\" VARCHAR, \"Flight_Number_Reporting_Airline\" BIGINT, \"OriginAirportID\" BIGINT, \"OriginAirportSeqID\" BIGINT, \"OriginCityMarketID\" BIGINT, \"Origin\" VARCHAR, \"OriginCityName\" VARCHAR, \"OriginState\" VARCHAR, \"OriginStateFips\" BIGINT, \"OriginStateName\" VARCHAR, \"OriginWac\" BIGINT, \"DestAirportID\" BIGINT, \"DestAirportSeqID\" BIGINT, \"DestCityMarketID\" BIGINT, \"Dest\" VARCHAR, \"DestCityName\" VARCHAR, \"DestState\" VARCHAR, \"DestStateFips\" BIGINT, \"DestStateName\" VARCHAR, \"DestWac\" BIGINT, \"CRSDepTime\" BIGINT, \"DepTime\" BIGINT, \"DepDelay\" BIGINT, \"DepDelayMinutes\" BIGINT, \"DepDel15\" BIGINT, \"DepartureDelayGroups\" BIGINT, \"DepTimeBlk\" VARCHAR, \"TaxiOut\" BIGINT, \"WheelsOff\" BIGINT, \"WheelsOn\" BIGINT, \"TaxiIn\" BIGINT, \"CRSArrTime\" BIGINT, \"ArrTime\" BIGINT, \"ArrDelay\" BIGINT, \"ArrDelayMinutes\" BIGINT, \"ArrDel15\" BIGINT, \"ArrivalDelayGroups\" BIGINT, \"ArrTimeBlk\" VARCHAR, \"Cancelled\" BIGINT, \"CancellationCode\" VARCHAR, \"Diverted\" BIGINT, \"CRSElapsedTime\" BIGINT, \"ActualElapsedTime\" BIGINT, \"AirTime\" BIGINT, \"Flights\" BIGINT, \"Distance\" BIGINT, \"DistanceGroup\" BIGINT, \"CarrierDelay\" BIGINT, \"WeatherDelay\" BIGINT, \"NASDelay\" BIGINT, \"SecurityDelay\" BIGINT, \"LateAircraftDelay\" BIGINT, \"FirstDepTime\" VARCHAR, \"TotalAddGTime\" VARCHAR, \"LongestAddGTime\" VARCHAR, \"DivAirportLandings\" VARCHAR, \"DivReachedDest\" VARCHAR, \"DivActualElapsedTime\" VARCHAR, \"DivArrDelay\" VARCHAR, \"DivDistance\" VARCHAR, \"Div1Airport\" VARCHAR, \"Div1AirportID\" VARCHAR, \"Div1AirportSeqID\" VARCHAR, \"Div1WheelsOn\" VARCHAR, \"Div1TotalGTime\" VARCHAR, \"Div1LongestGTime\" VARCHAR, \"Div1WheelsOff\" VARCHAR, \"Div1TailNum\" VARCHAR, \"Div2Airport\" VARCHAR, \"Div2AirportID\" VARCHAR, \"Div2AirportSeqID\" VARCHAR, \"Div2WheelsOn\" VARCHAR, \"Div2TotalGTime\" VARCHAR, \"Div2LongestGTime\" VARCHAR, \"Div2WheelsOff\" VARCHAR, \"Div2TailNum\" VARCHAR, \"Div3Airport\" VARCHAR, \"Div3AirportID\" VARCHAR, \"Div3AirportSeqID\" VARCHAR, \"Div3WheelsOn\" VARCHAR, \"Div3TotalGTime\" VARCHAR, \"Div3LongestGTime\" VARCHAR, \"Div3WheelsOff\" VARCHAR, \"Div3TailNum\" VARCHAR, \"Div4Airport\" VARCHAR, \"Div4AirportID\" VARCHAR, \"Div4AirportSeqID\" VARCHAR, \"Div4WheelsOn\" VARCHAR, \"Div4TotalGTime\" VARCHAR, \"Div4LongestGTime\" VARCHAR, \"Div4WheelsOff\" VARCHAR, \"Div4TailNum\" VARCHAR, \"Div5Airport\" VARCHAR, \"Div5AirportID\" VARCHAR, \"Div5AirportSeqID\" VARCHAR, \"Div5WheelsOn\" VARCHAR, \"Div5TotalGTime\" VARCHAR, \"Div5LongestGTime\" VARCHAR, \"Div5WheelsOff\" VARCHAR, \"Div5TailNum\" VARCHAR, \"Unnamed: 109\" VARCHAR))\n",
    "SELECT\n",
    "  TIME_PARSE(\"depaturetime\") AS \"__time\",\n",
    "  \"arrivalime\", \"Year\", \"Quarter\", \"Month\", \"DayofMonth\", \"DayOfWeek\", \"FlightDate\", \"Reporting_Airline\", \"DOT_ID_Reporting_Airline\",\n",
    "  \"IATA_CODE_Reporting_Airline\", \"Tail_Number\", \"Flight_Number_Reporting_Airline\", \n",
    "  \"OriginAirportID\", \"OriginAirportSeqID\", \"OriginCityMarketID\", \"Origin\", \"OriginCityName\", \"OriginState\", \"OriginStateFips\", \"OriginStateName\", \"OriginWac\",  \n",
    "  \"DestAirportID\", \"DestAirportSeqID\", \"DestCityMarketID\", \"Dest\", \"DestCityName\", \"DestState\", \"DestStateFips\", \"DestStateName\", \"DestWac\", \n",
    "  \"CRSDepTime\", \"DepTime\", \"DepDelay\", \"DepDelayMinutes\", \"DepDel15\", \"DepartureDelayGroups\", \"DepTimeBlk\",\n",
    "  \"TaxiOut\", \"WheelsOff\", \"WheelsOn\", \"TaxiIn\", \"CRSArrTime\", \"ArrTime\", \"ArrDelay\", \"ArrDelayMinutes\", \"ArrDel15\", \"ArrivalDelayGroups\", \"ArrTimeBlk\",\n",
    "  \"Cancelled\", \"CancellationCode\", \"Diverted\", \"CRSElapsedTime\", \"ActualElapsedTime\", \"AirTime\", \"Flights\", \"Distance\", \"DistanceGroup\", \"CarrierDelay\",\n",
    "  \"WeatherDelay\", \"NASDelay\", \"SecurityDelay\", \"LateAircraftDelay\", \"FirstDepTime\", \"TotalAddGTime\", \"LongestAddGTime\", \"DivAirportLandings\", \"DivReachedDest\",\n",
    "  \"DivActualElapsedTime\",  \"DivArrDelay\",  \"DivDistance\",  \"Div1Airport\",  \"Div1AirportID\",  \"Div1AirportSeqID\",  \"Div1WheelsOn\",  \"Div1TotalGTime\",\n",
    "  \"Div1LongestGTime\",  \"Div1WheelsOff\",  \"Div1TailNum\",  \"Div2Airport\",  \"Div2AirportID\",  \"Div2AirportSeqID\",  \"Div2WheelsOn\",  \"Div2TotalGTime\",\n",
    "  \"Div2LongestGTime\",  \"Div2WheelsOff\",  \"Div2TailNum\",  \"Div3Airport\",  \"Div3AirportID\",  \"Div3AirportSeqID\",  \"Div3WheelsOn\",  \"Div3TotalGTime\",\n",
    "  \"Div3LongestGTime\",  \"Div3WheelsOff\",  \"Div3TailNum\",  \"Div4Airport\",  \"Div4AirportID\",  \"Div4AirportSeqID\",  \"Div4WheelsOn\",  \"Div4TotalGTime\",\n",
    "  \"Div4LongestGTime\",  \"Div4WheelsOff\",  \"Div4TailNum\",  \"Div5Airport\",  \"Div5AirportID\",  \"Div5AirportSeqID\",  \"Div5WheelsOn\",  \"Div5TotalGTime\",\n",
    "  \"Div5LongestGTime\",  \"Div5WheelsOff\",  \"Div5TailNum\"\n",
    "FROM \"ext\"\n",
    "PARTITIONED BY YEAR'''\n",
    "\n",
    "display.run_task(sql)\n",
    "sql_client.wait_until_ready('flights-year')"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "36f0ddff-8a25-4c47-afe3-fec08066e11a",
   "metadata": {},
   "source": [
    "Here's the link to view the segments for this one on [the Apache Druid console in the Segments view](http://localhost:8888/unified-console.html#segments/datasource~flights-year).\n",
    "\n",
    "Notice that now you have all the rows in a single segment and that it has over 500,000 rows. Normally this would still be a small segment, but our target for this discussion is 50,000. This solution puts us at 10x.\n",
    "![](assets/segments-yearly.png)\n",
    "\n",
    "\n",
    "### Almost there...using `rowsPerSegment` to control segment size\n",
    "Given that there is a single month of data in this example, MONTH will not be any different than YEAR, DAY will be too granular as it will be 30x smaller and you need about 10x. You could try to use `P3D` to group segments into 3 day intervals and achieve our target. \n",
    "\n",
    "You can also just alter the execution parameters to force the partitioning of the large segment into the desired size by essentially cutting it into multiple segments 50,000 rows at a time, this is done with the query context parameter `rowsPerSegment`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9fe40c50-8cd7-4677-bd68-e846c1dbd749",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql='''\n",
    "REPLACE INTO \"flights-year-50k\" OVERWRITE ALL\n",
    "WITH \"ext\" AS (SELECT *\n",
    "FROM TABLE(\n",
    "  EXTERN(\n",
    "    '{\"type\":\"http\",\"uris\":[\"https://static.imply.io/example-data/flight_on_time/flights/On_Time_Reporting_Carrier_On_Time_Performance_(1987_present)_2005_11.csv.zip\"]}',\n",
    "    '{\"type\":\"csv\",\"findColumnsFromHeader\":true}'\n",
    "  )\n",
    ") EXTEND (\"depaturetime\" VARCHAR, \"arrivalime\" VARCHAR, \"Year\" BIGINT, \"Quarter\" BIGINT, \"Month\" BIGINT, \"DayofMonth\" BIGINT, \"DayOfWeek\" BIGINT, \"FlightDate\" VARCHAR, \"Reporting_Airline\" VARCHAR, \"DOT_ID_Reporting_Airline\" BIGINT, \"IATA_CODE_Reporting_Airline\" VARCHAR, \"Tail_Number\" VARCHAR, \"Flight_Number_Reporting_Airline\" BIGINT, \"OriginAirportID\" BIGINT, \"OriginAirportSeqID\" BIGINT, \"OriginCityMarketID\" BIGINT, \"Origin\" VARCHAR, \"OriginCityName\" VARCHAR, \"OriginState\" VARCHAR, \"OriginStateFips\" BIGINT, \"OriginStateName\" VARCHAR, \"OriginWac\" BIGINT, \"DestAirportID\" BIGINT, \"DestAirportSeqID\" BIGINT, \"DestCityMarketID\" BIGINT, \"Dest\" VARCHAR, \"DestCityName\" VARCHAR, \"DestState\" VARCHAR, \"DestStateFips\" BIGINT, \"DestStateName\" VARCHAR, \"DestWac\" BIGINT, \"CRSDepTime\" BIGINT, \"DepTime\" BIGINT, \"DepDelay\" BIGINT, \"DepDelayMinutes\" BIGINT, \"DepDel15\" BIGINT, \"DepartureDelayGroups\" BIGINT, \"DepTimeBlk\" VARCHAR, \"TaxiOut\" BIGINT, \"WheelsOff\" BIGINT, \"WheelsOn\" BIGINT, \"TaxiIn\" BIGINT, \"CRSArrTime\" BIGINT, \"ArrTime\" BIGINT, \"ArrDelay\" BIGINT, \"ArrDelayMinutes\" BIGINT, \"ArrDel15\" BIGINT, \"ArrivalDelayGroups\" BIGINT, \"ArrTimeBlk\" VARCHAR, \"Cancelled\" BIGINT, \"CancellationCode\" VARCHAR, \"Diverted\" BIGINT, \"CRSElapsedTime\" BIGINT, \"ActualElapsedTime\" BIGINT, \"AirTime\" BIGINT, \"Flights\" BIGINT, \"Distance\" BIGINT, \"DistanceGroup\" BIGINT, \"CarrierDelay\" BIGINT, \"WeatherDelay\" BIGINT, \"NASDelay\" BIGINT, \"SecurityDelay\" BIGINT, \"LateAircraftDelay\" BIGINT, \"FirstDepTime\" VARCHAR, \"TotalAddGTime\" VARCHAR, \"LongestAddGTime\" VARCHAR, \"DivAirportLandings\" VARCHAR, \"DivReachedDest\" VARCHAR, \"DivActualElapsedTime\" VARCHAR, \"DivArrDelay\" VARCHAR, \"DivDistance\" VARCHAR, \"Div1Airport\" VARCHAR, \"Div1AirportID\" VARCHAR, \"Div1AirportSeqID\" VARCHAR, \"Div1WheelsOn\" VARCHAR, \"Div1TotalGTime\" VARCHAR, \"Div1LongestGTime\" VARCHAR, \"Div1WheelsOff\" VARCHAR, \"Div1TailNum\" VARCHAR, \"Div2Airport\" VARCHAR, \"Div2AirportID\" VARCHAR, \"Div2AirportSeqID\" VARCHAR, \"Div2WheelsOn\" VARCHAR, \"Div2TotalGTime\" VARCHAR, \"Div2LongestGTime\" VARCHAR, \"Div2WheelsOff\" VARCHAR, \"Div2TailNum\" VARCHAR, \"Div3Airport\" VARCHAR, \"Div3AirportID\" VARCHAR, \"Div3AirportSeqID\" VARCHAR, \"Div3WheelsOn\" VARCHAR, \"Div3TotalGTime\" VARCHAR, \"Div3LongestGTime\" VARCHAR, \"Div3WheelsOff\" VARCHAR, \"Div3TailNum\" VARCHAR, \"Div4Airport\" VARCHAR, \"Div4AirportID\" VARCHAR, \"Div4AirportSeqID\" VARCHAR, \"Div4WheelsOn\" VARCHAR, \"Div4TotalGTime\" VARCHAR, \"Div4LongestGTime\" VARCHAR, \"Div4WheelsOff\" VARCHAR, \"Div4TailNum\" VARCHAR, \"Div5Airport\" VARCHAR, \"Div5AirportID\" VARCHAR, \"Div5AirportSeqID\" VARCHAR, \"Div5WheelsOn\" VARCHAR, \"Div5TotalGTime\" VARCHAR, \"Div5LongestGTime\" VARCHAR, \"Div5WheelsOff\" VARCHAR, \"Div5TailNum\" VARCHAR, \"Unnamed: 109\" VARCHAR))\n",
    "SELECT\n",
    "  TIME_PARSE(\"depaturetime\") AS \"__time\",\n",
    "  \"arrivalime\", \"Year\", \"Quarter\", \"Month\", \"DayofMonth\", \"DayOfWeek\", \"FlightDate\", \"Reporting_Airline\", \"DOT_ID_Reporting_Airline\",\n",
    "  \"IATA_CODE_Reporting_Airline\", \"Tail_Number\", \"Flight_Number_Reporting_Airline\", \n",
    "  \"OriginAirportID\", \"OriginAirportSeqID\", \"OriginCityMarketID\", \"Origin\", \"OriginCityName\", \"OriginState\", \"OriginStateFips\", \"OriginStateName\", \"OriginWac\",  \n",
    "  \"DestAirportID\", \"DestAirportSeqID\", \"DestCityMarketID\", \"Dest\", \"DestCityName\", \"DestState\", \"DestStateFips\", \"DestStateName\", \"DestWac\", \n",
    "  \"CRSDepTime\", \"DepTime\", \"DepDelay\", \"DepDelayMinutes\", \"DepDel15\", \"DepartureDelayGroups\", \"DepTimeBlk\",\n",
    "  \"TaxiOut\", \"WheelsOff\", \"WheelsOn\", \"TaxiIn\", \"CRSArrTime\", \"ArrTime\", \"ArrDelay\", \"ArrDelayMinutes\", \"ArrDel15\", \"ArrivalDelayGroups\", \"ArrTimeBlk\",\n",
    "  \"Cancelled\", \"CancellationCode\", \"Diverted\", \"CRSElapsedTime\", \"ActualElapsedTime\", \"AirTime\", \"Flights\", \"Distance\", \"DistanceGroup\", \"CarrierDelay\",\n",
    "  \"WeatherDelay\", \"NASDelay\", \"SecurityDelay\", \"LateAircraftDelay\", \"FirstDepTime\", \"TotalAddGTime\", \"LongestAddGTime\", \"DivAirportLandings\", \"DivReachedDest\",\n",
    "  \"DivActualElapsedTime\",  \"DivArrDelay\",  \"DivDistance\",  \"Div1Airport\",  \"Div1AirportID\",  \"Div1AirportSeqID\",  \"Div1WheelsOn\",  \"Div1TotalGTime\",\n",
    "  \"Div1LongestGTime\",  \"Div1WheelsOff\",  \"Div1TailNum\",  \"Div2Airport\",  \"Div2AirportID\",  \"Div2AirportSeqID\",  \"Div2WheelsOn\",  \"Div2TotalGTime\",\n",
    "  \"Div2LongestGTime\",  \"Div2WheelsOff\",  \"Div2TailNum\",  \"Div3Airport\",  \"Div3AirportID\",  \"Div3AirportSeqID\",  \"Div3WheelsOn\",  \"Div3TotalGTime\",\n",
    "  \"Div3LongestGTime\",  \"Div3WheelsOff\",  \"Div3TailNum\",  \"Div4Airport\",  \"Div4AirportID\",  \"Div4AirportSeqID\",  \"Div4WheelsOn\",  \"Div4TotalGTime\",\n",
    "  \"Div4LongestGTime\",  \"Div4WheelsOff\",  \"Div4TailNum\",  \"Div5Airport\",  \"Div5AirportID\",  \"Div5AirportSeqID\",  \"Div5WheelsOn\",  \"Div5TotalGTime\",\n",
    "  \"Div5LongestGTime\",  \"Div5WheelsOff\",  \"Div5TailNum\"\n",
    "FROM \"ext\"\n",
    "PARTITIONED BY YEAR'''\n",
    "# use a request so that you can specify the query context\n",
    "req = sql_client.sql_request(sql)\n",
    "req.add_context(\"rowsPerSegment\", \"50000\")  \n",
    "display.run_task(req)\n",
    "sql_client.wait_until_ready('flights-year-50k')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f8d8b77c-bf5d-4947-aef3-fee3c24f0ed1",
   "metadata": {},
   "source": [
    "The [segment view for this table](http://localhost:8888/unified-console.html#segments/datasource~flights-year-50k) shows 12 segments that are very close to our ideal segment size. \n",
    "\n",
    "But let's say that the application you are building is for year long analytics for individual airlines and some airline to airline comparisons. So let's move on to clustering...\n",
    "\n",
    "## Use CLUSTERED BY to apply secondary partitioning \n",
    "\n",
    "As mentioned in the introduction, time intervals of a given segment granularity can be subdivided into many segments. In the example above, you achieved this but there is no logic to the partitioning, they were just cut into 12 equally sized segments that were as close to the 50k target as possible.  \n",
    "\n",
    "Queries on this data will be frequently filtered on `IATA_CODE_Reporting_Airline` in order to provide individual airline analytics. So instead of just splitting the segments by quantity of rows, you can reorganize the data such that you can improve pruning when filtering on a given airline or a few airlines.\n",
    "You just need to add `CLUSTERED BY IATA_CODE_Reporting_Airline` to the ingestion request:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "af23ccfe-30ed-4510-b7dd-ea8efed102ad",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql='''\n",
    "REPLACE INTO \"flights-year-IATA\" OVERWRITE ALL\n",
    "WITH \"ext\" AS (SELECT *\n",
    "FROM TABLE(\n",
    "  EXTERN(\n",
    "    '{\"type\":\"http\",\"uris\":[\"https://static.imply.io/example-data/flight_on_time/flights/On_Time_Reporting_Carrier_On_Time_Performance_(1987_present)_2005_11.csv.zip\"]}',\n",
    "    '{\"type\":\"csv\",\"findColumnsFromHeader\":true}'\n",
    "  )\n",
    ") EXTEND (\"depaturetime\" VARCHAR, \"arrivalime\" VARCHAR, \"Year\" BIGINT, \"Quarter\" BIGINT, \"Month\" BIGINT, \"DayofMonth\" BIGINT, \"DayOfWeek\" BIGINT, \"FlightDate\" VARCHAR, \"Reporting_Airline\" VARCHAR, \"DOT_ID_Reporting_Airline\" BIGINT, \"IATA_CODE_Reporting_Airline\" VARCHAR, \"Tail_Number\" VARCHAR, \"Flight_Number_Reporting_Airline\" BIGINT, \"OriginAirportID\" BIGINT, \"OriginAirportSeqID\" BIGINT, \"OriginCityMarketID\" BIGINT, \"Origin\" VARCHAR, \"OriginCityName\" VARCHAR, \"OriginState\" VARCHAR, \"OriginStateFips\" BIGINT, \"OriginStateName\" VARCHAR, \"OriginWac\" BIGINT, \"DestAirportID\" BIGINT, \"DestAirportSeqID\" BIGINT, \"DestCityMarketID\" BIGINT, \"Dest\" VARCHAR, \"DestCityName\" VARCHAR, \"DestState\" VARCHAR, \"DestStateFips\" BIGINT, \"DestStateName\" VARCHAR, \"DestWac\" BIGINT, \"CRSDepTime\" BIGINT, \"DepTime\" BIGINT, \"DepDelay\" BIGINT, \"DepDelayMinutes\" BIGINT, \"DepDel15\" BIGINT, \"DepartureDelayGroups\" BIGINT, \"DepTimeBlk\" VARCHAR, \"TaxiOut\" BIGINT, \"WheelsOff\" BIGINT, \"WheelsOn\" BIGINT, \"TaxiIn\" BIGINT, \"CRSArrTime\" BIGINT, \"ArrTime\" BIGINT, \"ArrDelay\" BIGINT, \"ArrDelayMinutes\" BIGINT, \"ArrDel15\" BIGINT, \"ArrivalDelayGroups\" BIGINT, \"ArrTimeBlk\" VARCHAR, \"Cancelled\" BIGINT, \"CancellationCode\" VARCHAR, \"Diverted\" BIGINT, \"CRSElapsedTime\" BIGINT, \"ActualElapsedTime\" BIGINT, \"AirTime\" BIGINT, \"Flights\" BIGINT, \"Distance\" BIGINT, \"DistanceGroup\" BIGINT, \"CarrierDelay\" BIGINT, \"WeatherDelay\" BIGINT, \"NASDelay\" BIGINT, \"SecurityDelay\" BIGINT, \"LateAircraftDelay\" BIGINT, \"FirstDepTime\" VARCHAR, \"TotalAddGTime\" VARCHAR, \"LongestAddGTime\" VARCHAR, \"DivAirportLandings\" VARCHAR, \"DivReachedDest\" VARCHAR, \"DivActualElapsedTime\" VARCHAR, \"DivArrDelay\" VARCHAR, \"DivDistance\" VARCHAR, \"Div1Airport\" VARCHAR, \"Div1AirportID\" VARCHAR, \"Div1AirportSeqID\" VARCHAR, \"Div1WheelsOn\" VARCHAR, \"Div1TotalGTime\" VARCHAR, \"Div1LongestGTime\" VARCHAR, \"Div1WheelsOff\" VARCHAR, \"Div1TailNum\" VARCHAR, \"Div2Airport\" VARCHAR, \"Div2AirportID\" VARCHAR, \"Div2AirportSeqID\" VARCHAR, \"Div2WheelsOn\" VARCHAR, \"Div2TotalGTime\" VARCHAR, \"Div2LongestGTime\" VARCHAR, \"Div2WheelsOff\" VARCHAR, \"Div2TailNum\" VARCHAR, \"Div3Airport\" VARCHAR, \"Div3AirportID\" VARCHAR, \"Div3AirportSeqID\" VARCHAR, \"Div3WheelsOn\" VARCHAR, \"Div3TotalGTime\" VARCHAR, \"Div3LongestGTime\" VARCHAR, \"Div3WheelsOff\" VARCHAR, \"Div3TailNum\" VARCHAR, \"Div4Airport\" VARCHAR, \"Div4AirportID\" VARCHAR, \"Div4AirportSeqID\" VARCHAR, \"Div4WheelsOn\" VARCHAR, \"Div4TotalGTime\" VARCHAR, \"Div4LongestGTime\" VARCHAR, \"Div4WheelsOff\" VARCHAR, \"Div4TailNum\" VARCHAR, \"Div5Airport\" VARCHAR, \"Div5AirportID\" VARCHAR, \"Div5AirportSeqID\" VARCHAR, \"Div5WheelsOn\" VARCHAR, \"Div5TotalGTime\" VARCHAR, \"Div5LongestGTime\" VARCHAR, \"Div5WheelsOff\" VARCHAR, \"Div5TailNum\" VARCHAR, \"Unnamed: 109\" VARCHAR))\n",
    "SELECT\n",
    "  TIME_PARSE(\"depaturetime\") AS \"__time\",\n",
    "  \"arrivalime\", \"Year\", \"Quarter\", \"Month\", \"DayofMonth\", \"DayOfWeek\", \"FlightDate\", \"Reporting_Airline\", \"DOT_ID_Reporting_Airline\",\n",
    "  \"IATA_CODE_Reporting_Airline\", \"Tail_Number\", \"Flight_Number_Reporting_Airline\", \n",
    "  \"OriginAirportID\", \"OriginAirportSeqID\", \"OriginCityMarketID\", \"Origin\", \"OriginCityName\", \"OriginState\", \"OriginStateFips\", \"OriginStateName\", \"OriginWac\",  \n",
    "  \"DestAirportID\", \"DestAirportSeqID\", \"DestCityMarketID\", \"Dest\", \"DestCityName\", \"DestState\", \"DestStateFips\", \"DestStateName\", \"DestWac\", \n",
    "  \"CRSDepTime\", \"DepTime\", \"DepDelay\", \"DepDelayMinutes\", \"DepDel15\", \"DepartureDelayGroups\", \"DepTimeBlk\",\n",
    "  \"TaxiOut\", \"WheelsOff\", \"WheelsOn\", \"TaxiIn\", \"CRSArrTime\", \"ArrTime\", \"ArrDelay\", \"ArrDelayMinutes\", \"ArrDel15\", \"ArrivalDelayGroups\", \"ArrTimeBlk\",\n",
    "  \"Cancelled\", \"CancellationCode\", \"Diverted\", \"CRSElapsedTime\", \"ActualElapsedTime\", \"AirTime\", \"Flights\", \"Distance\", \"DistanceGroup\", \"CarrierDelay\",\n",
    "  \"WeatherDelay\", \"NASDelay\", \"SecurityDelay\", \"LateAircraftDelay\", \"FirstDepTime\", \"TotalAddGTime\", \"LongestAddGTime\", \"DivAirportLandings\", \"DivReachedDest\",\n",
    "  \"DivActualElapsedTime\",  \"DivArrDelay\",  \"DivDistance\",  \"Div1Airport\",  \"Div1AirportID\",  \"Div1AirportSeqID\",  \"Div1WheelsOn\",  \"Div1TotalGTime\",\n",
    "  \"Div1LongestGTime\",  \"Div1WheelsOff\",  \"Div1TailNum\",  \"Div2Airport\",  \"Div2AirportID\",  \"Div2AirportSeqID\",  \"Div2WheelsOn\",  \"Div2TotalGTime\",\n",
    "  \"Div2LongestGTime\",  \"Div2WheelsOff\",  \"Div2TailNum\",  \"Div3Airport\",  \"Div3AirportID\",  \"Div3AirportSeqID\",  \"Div3WheelsOn\",  \"Div3TotalGTime\",\n",
    "  \"Div3LongestGTime\",  \"Div3WheelsOff\",  \"Div3TailNum\",  \"Div4Airport\",  \"Div4AirportID\",  \"Div4AirportSeqID\",  \"Div4WheelsOn\",  \"Div4TotalGTime\",\n",
    "  \"Div4LongestGTime\",  \"Div4WheelsOff\",  \"Div4TailNum\",  \"Div5Airport\",  \"Div5AirportID\",  \"Div5AirportSeqID\",  \"Div5WheelsOn\",  \"Div5TotalGTime\",\n",
    "  \"Div5LongestGTime\",  \"Div5WheelsOff\",  \"Div5TailNum\"\n",
    "FROM \"ext\"\n",
    "PARTITIONED BY YEAR\n",
    "CLUSTERED BY IATA_CODE_Reporting_Airline\n",
    "'''\n",
    "# use a request so that you can specify the query context\n",
    "req = sql_client.sql_request(sql)\n",
    "req.add_context(\"rowsPerSegment\", \"50000\")  # you still want to target the ideal 50k per segment, normally greater than 3 million \n",
    "display.run_task(req)\n",
    "sql_client.wait_until_ready('flights-year-IATA')"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "8799d234-c2f3-4277-9661-8931b94d58ff",
   "metadata": {},
   "source": [
    "The [segment view for this table](http://localhost:8888/unified-console.html#segments/datasource~flights-year-IATA) shows 12 segments that are still very close to our ideal segment size, but now they have some new metadata based on the clustering columns in the `Shard Spec` column. It shows which range of values of the IATA_CODE_Reporting_Airline column are available in the segment. This is very useful for pruning when filtering on that column because Druid can prune to 1 or 2 segment files when looking for a single airline.\n",
    "\n",
    "![](assets/segments-iata.png)\n",
    "\n",
    "## Measuring the impact of segment layout on query performance\n",
    "In this part of the notebook, now that the data has been loaded using a few different partitioning strategies, you will compare performance of each approach.\n",
    "The `useCache` query parameter will be set to false for the purposes of this part of the notebook so that can measure performance of the same query with each table.\n",
    "\n",
    "### The query\n",
    "It is well known that flight traffic follows a weekly pattern as well as a seasonal pattern. Given that you only have a month, let's take a look at the maximum and average delay for a single airline by day of the week so that you can see the pattern and how it affects delays. You'll only consider the departure delay for now and only use delays greater than 2 minutes as \"delayed\".\n",
    "\n",
    "The following cell defines a function you can use to measure performance:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "77a71e6a-73a2-4bd7-9e8d-fd7cb966da65",
   "metadata": {},
   "outputs": [],
   "source": [
    "# The measure_query function runs the specified sql query multiple times \n",
    "# and measures the duration of each run.\n",
    "# The output is a string with the results including the mean, minimum,\n",
    "# and maximum durations measured.\n",
    "def measure_query( sql: str, iterations: int ):\n",
    "    req = sql_client.sql_request(sql)\n",
    "    req.add_context(\"populateCache\", \"false\")  # run without caching results to get a real sense of performance\n",
    "    req.add_context(\"useCache\", \"false\")  # do not use cached results\n",
    "    stats = []\n",
    "    while (iterations>0):\n",
    "      start = datetime.now()\n",
    "      sql_client.sql(req)\n",
    "      end = datetime.now()\n",
    "      stats.append( (end - start).total_seconds() * 10**3 ) # add run time in milliseconds\n",
    "      iterations -=1\n",
    "    return f\"Results = avg:{mean(stats)} ms   min:{min(stats)} ms  max:{max(stats)} ms\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "de7c2156-99bf-42bf-858e-537209c20fab",
   "metadata": {},
   "source": [
    "Run the test SQL to see that in the month of November in 2005, American Airlines had the highest percentage of delayed flights on Sundays(7) although the worst average delay occurs on Fridays(5):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "42569b72-e018-4576-b8e5-121d2e2a36c5",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = '''\n",
    "SELECT EXTRACT( DOW FROM __time) as day_of_week, \n",
    "       AVG(DepDelay) FILTER ( WHERE DepDelay>120) as avg_delay,\n",
    "       MAX(DepDelay) FILTER ( WHERE DepDelay>120) as max_delay,\n",
    "       ROUND(COUNT(1) FILTER ( WHERE DepDelay>120) * 100.0 / COUNT(1), 1) as percent_delayed\n",
    "FROM \"{}\"\n",
    "WHERE \"IATA_CODE_Reporting_Airline\" = 'AA'\n",
    "GROUP BY 1 \n",
    "ORDER BY 1 ASC\n",
    "'''\n",
    "\n",
    "display.sql(sql.format(\"flights-hour\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d83c7eb1-6d0a-4a08-aa1b-03ec8b755539",
   "metadata": {},
   "source": [
    "### Performance results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "dc4c2524-0eba-4bc6-84ed-da3a25aa5fbe",
   "metadata": {},
   "outputs": [],
   "source": [
    "# run 100 queries with each table\n",
    "use_table=\"flights-hour\"\n",
    "print (f\"{use_table} {measure_query(sql.format(use_table), 100)}\")\n",
    "\n",
    "use_table=\"flights-year\"\n",
    "print (f\"{use_table} {measure_query(sql.format(use_table), 100)}\")\n",
    "\n",
    "use_table=\"flights-year-50k\"\n",
    "print (f\"{use_table} {measure_query(sql.format(use_table), 100)}\")\n",
    "\n",
    "use_table=\"flights-year-IATA\"\n",
    "print (f\"{use_table} {measure_query(sql.format(use_table), 100)}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d0716364-af1f-4eb1-9e75-3ce3c5617344",
   "metadata": {},
   "source": [
    "### Single concurrency results\n",
    "\n",
    "Given that the characteristics of your laptop or other execution environment are likely different than the `Apple M2 Pro` that this was built on, your results will vary, my results were:\n",
    "```\n",
    "flights-hour Results      = avg:28.848 ms   min:25.198 ms   max:64.379 ms\n",
    "flights-year Results      = avg:15.451 ms   min:13.891 ms   max:17.940 ms\n",
    "flights-year-50k Results  = avg:12.844 ms   min:11.594 ms   max:15.130 ms\n",
    "flights-year-IATA Results = avg:12.574 ms   min:11.221 ms   max:15.204 ms\n",
    "```\n",
    "\n",
    "- `flights-hour` - With segment granularity of hour, there are approximately 30*24= 720 segment files to scan, there is no pruning, so even though the segments are tiny, each one is examined separately and the results are then merged, so doing that 720 times makes this the slowest option.\n",
    "- `flights-year` - Here you are using a single segment, it is much larger at 500k+ rows, but Druid can still use the indexes associated to all dimension columns to process this pretty fast and it only does it once, so this is faster then hourly. This shows how the number of segments needed for a query is a factor for performance. Less segments to process tends to improve performance, but not always.\n",
    "- `flights-year-50k` - Now each query had to process twelve 50k segments, but given that the historical does this concurrently on multiple threads, the parallelism is enough to improve over the single 500k segment of the `flights-year` table. \n",
    "- `flights-year-IATA` - Given the shard spec of the 12 resulting segments, you can see that there are rows with IATA Code \"AA\" in 2 of the segments. They are both processed in parallel and there are less segments to process, so this results in the fastest. \n",
    "\n",
    "### Effects of higher concurrency\n",
    "\n",
    "Above, the results for `flights-year-IATA` seem very similar to the `flights-year-50k` results, but a real system processes multiple queries at once. Expanding the test to run with higher concurrency will tell us more. The expectation is that the organization of the data that requires less work per query will ultimately be better for higher concurrency.\n",
    "\n",
    "You'll need another set of functions to drive the queries in parallel.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "03175503-1a3f-4a50-8051-c97b084a6f6f",
   "metadata": {},
   "outputs": [],
   "source": [
    "from threading import Thread\n",
    "from datetime import datetime\n",
    "from statistics import mean, median\n",
    "\n",
    "\n",
    "# Custom thread class that runs a query multiple times so that you can execute concurrently on multiple threads\n",
    "class QueryThread(Thread):\n",
    "    # constructor\n",
    "    def __init__(self, sql: str, iterations: int ):\n",
    "        # execute the base constructor\n",
    "        Thread.__init__(self)\n",
    "        # set a default value\n",
    "        self.stats = []\n",
    "        self.sql = sql\n",
    "        self.iterations = iterations\n",
    " \n",
    "    # function executed in a new thread\n",
    "    def run(self):\n",
    "        self.stats = self.measure_query( self.sql, self.iterations)\n",
    "        \n",
    "    def measure_query( self, sql: str, iterations: int ):\n",
    "        req = sql_client.sql_request(sql)\n",
    "        req.add_context(\"populateCache\", \"false\")  # run without caching results to get a real sense of performance\n",
    "        req.add_context(\"useCache\", \"false\")  # do not use cached results\n",
    "        stats=[]\n",
    "        while (iterations>0):\n",
    "          start = datetime.now()\n",
    "          sql_client.sql(req)\n",
    "          end = datetime.now()\n",
    "          stats.append( (end - start).total_seconds() * 10**3 ) # add run time in milliseconds\n",
    "          iterations -=1\n",
    "        return stats\n",
    "\n",
    "# This function will spawns \"numThreads\" parallel threads that each run \n",
    "# the specified \"sql\" \"iterations\" times and measures overall results.\n",
    "def test_in_parallel( sql: str, numThreads: int, iterations: int):\n",
    "    threads=[]\n",
    "    while numThreads>0:\n",
    "        try:\n",
    "            thrd = QueryThread( sql, iterations)\n",
    "            thrd.start()\n",
    "            threads.append(thrd)\n",
    "        except:\n",
    "           print (\"Error: unable to start thread\")\n",
    "        numThreads-=1\n",
    "    total_stats = []\n",
    "    for thrd in threads:\n",
    "        thrd.join()\n",
    "        total_stats.extend(thrd.stats)\n",
    "    return f\"Results = count:{len(total_stats)}  avg:{median(total_stats)} ms   min:{min(total_stats)} ms  max:{max(total_stats)} ms\"  "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d128e713-5a1f-48d0-9dde-8028861e9f19",
   "metadata": {},
   "source": [
    "#### Concurrency = 2\n",
    "In this test you will use the best table designs from the previous step (`flights-year-50k`, `flights-year-IATA`) to process queries in two concurrent threads."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f85b8419-936c-4f46-b9ab-869ee8779c7c",
   "metadata": {},
   "outputs": [],
   "source": [
    "use_table=\"flights-year-50k\"\n",
    "print (f\"{use_table} {test_in_parallel(sql.format(use_table), 2, 50)}\")\n",
    "\n",
    "use_table=\"flights-year-IATA\"\n",
    "print (f\"{use_table} {test_in_parallel(sql.format(use_table), 2, 50)}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b7d2a104-7d69-4fa2-840d-ab6787143658",
   "metadata": {},
   "source": [
    "My results:\n",
    "```\n",
    "flights-year-50k Results  = count:100  avg:13.481 ms   min:11.139 ms  max:38.034 ms\n",
    "flights-year-IATA Results = count:100  avg:13.073 ms   min:11.251 ms  max:20.746 ms\n",
    "```\n",
    "\n",
    "The avg suffered a bit for both tables when compared to the single concurrency test. \n",
    "The interesting change is in the max where it is almost `1.9` x faster on the IATA clustered table.\n",
    "Let's take it up another notch...\n",
    "\n",
    "#### Concurrency = 4\n",
    "Double the concurrency and re-run the test:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "692ab885-587f-434a-a69e-4cf628c2f553",
   "metadata": {},
   "outputs": [],
   "source": [
    "use_table=\"flights-year-50k\"\n",
    "print (f\"{use_table} {test_in_parallel(sql.format(use_table), 4, 50)}\")\n",
    "\n",
    "use_table=\"flights-year-IATA\"\n",
    "print (f\"{use_table} {test_in_parallel(sql.format(use_table), 4, 50)}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ddaf1e58-ba7a-49e2-8cda-60fc52979d6f",
   "metadata": {},
   "source": [
    "My results:\n",
    "```\n",
    "flights-year-50k Results  = count:200  avg:14.543 ms   min:11.341 ms  max:68.145 ms\n",
    "flights-year-IATA Results = count:200  avg:13.909 ms   min:11.019 ms  max:33.341 ms\n",
    "```\n",
    "The avg continues to grow, meaning that you've likely hit some limit in resources. The minimum continues unchanged which also makes sense because the first time the query runs it probably doesn't need to wait. Subsequent parallel requests will need to wait in a queue while resources are released, as you increase the concurrency, there's more queries in the queue.\n",
    "\n",
    "The important difference continues to be in the max value that is now `2.06` x faster on the IATA table.\n",
    "Let's get one more data point...\n",
    "\n",
    "#### Concurrency = 8"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ec00d899-7f25-4254-ab23-b79a7920e9e3",
   "metadata": {},
   "outputs": [],
   "source": [
    "use_table=\"flights-year-50k\"\n",
    "print (f\"{use_table} {test_in_parallel(sql.format(use_table), 8, 50)}\")\n",
    "\n",
    "use_table=\"flights-year-IATA\"\n",
    "print (f\"{use_table} {test_in_parallel(sql.format(use_table), 8, 50)}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f36ff06e-e2ab-4a7a-8c08-c9966b336347",
   "metadata": {
    "jp-MarkdownHeadingCollapsed": true
   },
   "source": [
    "My results:\n",
    "```\n",
    "flights-year-50k Results  = count:400  avg:25.969 ms   min:11.595 ms  max:97.475 ms\n",
    "flights-year-IATA Results = count:400  avg:24.171 ms   min:11.979 ms  max:43.507 ms\n",
    "```\n",
    "Again, the avg continues to grow and the minimum continues unchanged.\n",
    "\n",
    "The important difference continues to be in the max value that is now `2.25` x faster on the IATA table.\n",
    "\n",
    "As you increase concurrency the IATA table continues to be more effective at dealing with concurrency. This effect will be more noticeable in a clustered deployment with more independent resources and as segments sizes grow to their more normal 3-10 million rows each."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "44738d6d-cec2-40ad-aaba-998c758c63f4",
   "metadata": {},
   "source": [
    "## Clean up\n",
    "\n",
    "Run the following cell to remove the tables created in the database."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8082b545-ba7f-4ede-bb6e-2a6dd62ba0d8",
   "metadata": {},
   "outputs": [],
   "source": [
    "druid.datasources.drop(\"flights-hour\")\n",
    "druid.datasources.drop(\"flights-year\")\n",
    "druid.datasources.drop(\"flights-year-50k\")\n",
    "druid.datasources.drop(\"flights-year-IATA\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "54b8d5fe-ba85-4b5b-9669-0dd47dfbccd1",
   "metadata": {},
   "source": [
    "## Summary\n",
    "\n",
    "* You learned that\n",
    "  * Druid works best with fewer optimally sized segments.\n",
    "  * Optimal segment size will vary based on the use case, but 5 million rows is a good starting point.\n",
    "  * PARTITIONED BY is used to define the segment granularity of a table.\n",
    "  * CLUSTERED BY is used to organize segments within a segment granularity time interval.\n",
    "  * CLUSTERED BY columns should be selected based on the most common filter criteria for the queries it will serve.\n",
    "  \n",
    "## Learn more\n",
    "\n",
    "* Test performance of different test queries against each of the partitioning strategies.\n",
    "* Take a look at the [06-partitioning-while-streaming.ipynb](06-partitioning-while-streaming.ipynb) notebook which takes a look at how to address partitioning and clustering for streaming.\n",
    "* Read about:\n",
    "  * [SQL Based Ingestion](https://druid.apache.org/docs/latest/multi-stage-query/)\n",
    "  * [Partitioning](https://druid.apache.org/docs/latest/ingestion/partitioning)\n",
    "  * [PARTITIONED BY and CLUSTERED BY docs](https://druid.apache.org/docs/latest/ingestion/partitioning)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7283791c-ea7f-4649-aeee-a4b31d829916",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "execution": {
   "allow_errors": true,
   "timeout": 300
  },
  "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.11.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}