python/notebooks/druid_ingestion_test.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "#\n",
    "# Copyright 2021 Rovio Entertainment Corporation\n",
    "#\n",
    "# Licensed under the Apache License, Version 2.0 (the \"License\");\n",
    "# you may not use this file except in compliance with the License.\n",
    "# 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, software\n",
    "# distributed under the License is distributed on an \"AS IS\" BASIS,\n",
    "# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n",
    "# See the License for the specific language governing permissions and\n",
    "# limitations under the License.\n",
    "#"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "pycharm": {
     "name": "#%% md\n"
    }
   },
   "source": [
    "# PRE-REQUISITES\n",
    "\n",
    "        AWS_PROFILE=smoke\n",
    "        JAR_BUCKET=<REPLACE THIS>\n",
    "\n",
    "## IF TESTING PYTHON CHANGES MANUALLY\n",
    "\n",
    "3. Build a zip of the python wrapper:\n",
    "\n",
    "        cd python \\\n",
    "          && zip --exclude='*.pyc' --exclude='*__pycache__*' --exclude='*~' --exclude='.pytest_cache' \\\n",
    "            -FSr ../target/rovio_ingest.zip rovio_ingest ; cd ..\n",
    "\n",
    "4. Copy the zip to s3:\n",
    "\n",
    "        aws s3 --profile $AWS_PROFILE cp \\\n",
    "          target/rovio_ingest.zip \\\n",
    "          s3://$JAR_BUCKET/tmp/juho/druid/python/rovio_ingest.zip\n",
    "\n",
    "Then invert the boolean in the cell below to use it in spark_conf.\n",
    "And skip the cell that would call install_pypi_package.\n",
    "\n",
    "## IF TESTING JAR CHANGES MANUALLY:\n",
    "\n",
    "1. Build the package (shaded jar) on command line:\n",
    "\n",
    "        mvn package -DskipTests\n",
    "\n",
    "2. A) Copy the shaded jar to s3:\n",
    "\n",
    "        aws s3 --profile $AWS_PROFILE cp \\\n",
    "          target/rovio-ingest-1.0.5_spark_3.0.1-SNAPSHOT.jar \\\n",
    "          s3://$JAR_BUCKET/tmp/juho/druid/jars/rovio-ingest-1.0.5_spark_3.0.1-SNAPSHOT.jar\n",
    "\n",
    "2. B) Copy the plain jar to s3: \n",
    "\n",
    "        aws s3 --profile $AWS_PROFILE cp \\\n",
    "          target/original-rovio-ingest-1.0.5_spark_3.0.1-SNAPSHOT.jar \\\n",
    "          s3://$JAR_BUCKET/tmp/juho/druid/jars/original-rovio-ingest-1.0.5_spark_3.0.1-SNAPSHOT.jar\n",
    "\n",
    "Then invert the boolean in the cell below to use it in spark_conf."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "%load_ext sparkmagic.magics"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "ENV = \"smoke\"\n",
    "PREFIX = \"tmp/juho/\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "import boto3\n",
    "\n",
    "ssm_client = boto3.session.Session(profile_name=ENV).client(service_name=\"ssm\")\n",
    "\n",
    "# secrets can be added at\n",
    "# https://console.aws.amazon.com/systems-manager/parameters/?region=us-east-1\n",
    "def get_param(secret_name: str) -> str:\n",
    "    return ssm_client.get_parameter(Name=\"/dataengineering/\" + secret_name)[\"Parameter\"][\"Value\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "import json\n",
    "from IPython import get_ipython\n",
    "\n",
    "def set_spark_config(conf_dict):\n",
    "    get_ipython().run_cell_magic('spark', 'config', json.dumps(conf_dict))\n",
    "\n",
    "def create_spark_session_with_host(host):\n",
    "    get_ipython().run_line_magic('spark', 'add -l python -u http://{}:8998'.format(host))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Starting Spark application\n"
     ]
    },
    {
     "data": {
      "text/plain": "<IPython.core.display.HTML object>",
      "text/html": "<table>\n<tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>Current session?</th></tr><tr><td>7</td><td>application_1673361870104_0145</td><td>pyspark</td><td>idle</td><td><a target=\"_blank\" href=\"http://ip-10-0-10-101.cloud-dev.rovio.com:20888/proxy/application_1673361870104_0145/\">Link</a></td><td><a target=\"_blank\" href=\"http://ip-10-0-10-20.cloud-dev.rovio.com:8042/node/containerlogs/container_1673361870104_0145_01_000001/livy\">Link</a></td><td>✔</td></tr></table>"
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…",
      "application/vnd.jupyter.widget-view+json": {
       "version_major": 2,
       "version_minor": 0,
       "model_id": "7636680cf91c4641af97375e42bc83b0"
      }
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SparkSession available as 'spark'.\n"
     ]
    }
   ],
   "source": [
    "packages_bucket = get_param(\"rovio-ingest/packages_bucket\")\n",
    "\n",
    "spark_conf = {\n",
    "  \"conf\": {\n",
    "    \"spark.hadoop.mapreduce.input.fileinputformat.input.dir.recursive\": \"true\",\n",
    "    \"spark.sql.hive.caseSensitiveInferenceMode\": \"NEVER_INFER\",\n",
    "    \"spark.pyspark.python\": \"python3\",\n",
    "    \"spark.sql.session.timeZone\": \"UTC\",\n",
    "    # alternative if using a snapshot version\n",
    "#    \"spark.jars.repositories\": \"https://s01.oss.sonatype.org/content/repositories/snapshots\",\n",
    "#    \"spark.jars.packages\": \"com.rovio.ingest:rovio-ingest:1.0.5_spark_3.0.1-SNAPSHOT\"\n",
    "    \"spark.jars.packages\": \"com.rovio.ingest:rovio-ingest:1.0.4_spark_3.0.1\"\n",
    "  }\n",
    "}\n",
    "\n",
    "# Assuming AWS EMR\n",
    "if True:\n",
    "    spark_conf[\"conf\"][\"spark.pyspark.python\"] = \"python3\"\n",
    "    spark_conf[\"conf\"][\"spark.pyspark.virtualenv.enabled\"] = \"true\"\n",
    "    spark_conf[\"conf\"][\"spark.pyspark.virtualenv.type\"] = \"native\"\n",
    "    spark_conf[\"conf\"][\"spark.pyspark.virtualenv.bin.path\"] = \"/usr/bin/virtualenv\"\n",
    "\n",
    "# Enable this to test with a manually built & copied zip instead of published package from PyPI\n",
    "if False:\n",
    "  spark_conf[\"conf\"][\"spark.submit.pyFiles\"] = \\\n",
    "    f\"s3://{packages_bucket}/{PREFIX}druid/python/rovio_ingest.zip\"\n",
    "\n",
    "# Enable this to test with a manually built & copied jar instead of published package from maven\n",
    "if False:\n",
    "  spark_conf[\"conf\"][\"spark.jars\"] = \\\n",
    "    f\"s3://{packages_bucket}/{PREFIX}druid/jars/rovio-ingest-1.0.5_spark_3.0.1-SNAPSHOT.jar\"\n",
    "  del spark_conf[\"conf\"][\"spark.jars.packages\"]\n",
    "\n",
    "set_spark_config(spark_conf)\n",
    "create_spark_session_with_host(get_param(\"spark3/shared/host\"))\n",
    "\n",
    "# to debug problems in session creation, see livy session logs at http://{host}:8998/ui"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…",
      "application/vnd.jupyter.widget-view+json": {
       "version_major": 2,
       "version_minor": 0,
       "model_id": "df3cf02fcfab45928622f38559736df6"
      }
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "%%spark\n",
    "\n",
    "# This extension is provided by AWS EMR\n",
    "# spark.sparkContext.list_packages()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Collecting rovio-ingest\n",
      "  Downloading https://files.pythonhosted.org/packages/e2/94/c268caf284b71b3a26f61b8f92abad35a89a18efff66f77d3424cda1ab7c/rovio_ingest-1.0.1-py3-none-any.whl\n",
      "Collecting pyspark<4.0.0,>=3.0.0 (from rovio-ingest)\n",
      "Collecting py4j==0.10.9 (from pyspark<4.0.0,>=3.0.0->rovio-ingest)\n",
      "  Using cached https://files.pythonhosted.org/packages/9e/b6/6a4fb90cd235dc8e265a6a2067f2a2c99f0d91787f06aca4bcf7c23f3f80/py4j-0.10.9-py2.py3-none-any.whl\n",
      "Installing collected packages: py4j, pyspark, rovio-ingest\n",
      "Successfully installed py4j-0.10.9 pyspark-3.1.1 rovio-ingest-1.0.1"
     ]
    }
   ],
   "source": [
    "%%spark\n",
    "\n",
    "# This extension is provided by AWS EMR.\n",
    "# If not on EMR:\n",
    "#    A) install the module with pip on the cluster before creating the spark session\n",
    "#    B) build a zip & use with spark.submit.pyFiles\n",
    "# Use the latest stable release from PyPI.\n",
    "spark.sparkContext.install_pypi_package(\"rovio-ingest\")\n",
    "# Use a specific version to install a pre-release from PyPI.\n",
    "#spark.sparkContext.install_pypi_package(\"rovio-ingest==0.0.1.dev14\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…",
      "application/vnd.jupyter.widget-view+json": {
       "version_major": 2,
       "version_minor": 0,
       "model_id": "41ee2e2ce4a5435086fc97100595e338"
      }
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "%%spark\n",
    "\n",
    "import boto3\n",
    "\n",
    "ssm_client = boto3.session.Session(region_name=\"us-east-1\").client(service_name=\"ssm\")\n",
    "\n",
    "def get_param(secret_name: str) -> str:\n",
    "    return ssm_client.get_parameter(Name=\"/dataengineering/\" + secret_name)[\"Parameter\"][\"Value\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…",
      "application/vnd.jupyter.widget-view+json": {
       "version_major": 2,
       "version_minor": 0,
       "model_id": "35c75531a1264a0f81c22afea7f3bc6e"
      }
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-------+----------+-------------------+\n",
      "|dau|revenue|    app_id|         event_date|\n",
      "+---+-------+----------+-------------------+\n",
      "|  5|   30.0|testclient|2019-10-01 00:00:00|\n",
      "|  2|   15.0|testclient|2019-10-02 00:00:00|\n",
      "+---+-------+----------+-------------------+"
     ]
    }
   ],
   "source": [
    "%%spark\n",
    "from datetime import datetime\n",
    "from pyspark.sql import functions as f, types as t, SparkSession\n",
    "\n",
    "spark: SparkSession = spark\n",
    "schema = 'dau:BIGINT, revenue:DOUBLE, app_id:STRING, event_date:TIMESTAMP'\n",
    "df = spark.createDataFrame([[5, 30.0, 'testclient', datetime(2019, 10, 1)],\n",
    "                            [2, 15.0, 'testclient', datetime(2019, 10, 2)]],\n",
    "                            schema)\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…",
      "application/vnd.jupyter.widget-view+json": {
       "version_major": 2,
       "version_minor": 0,
       "model_id": "4c65426dce6642c7b7cf6e1cab629484"
      }
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "== Parsed Logical Plan ==\n",
      "Project [dau#99L, revenue#100, app_id#101, event_date#102, __PARTITION_TIME__#130, __PARTITION_NUM__#147]\n",
      "+- RepartitionByExpression [__PARTITION_TIME__#130, __PARTITION_NUM__#147]\n",
      "   +- Project [dau#99L, revenue#100, app_id#101, event_date#102, __PARTITION_TIME__#130, __num_rows__#140, cast((cast((__num_rows__#140 - 1) as double) / cast(5000000 as double)) as int) AS __PARTITION_NUM__#147]\n",
      "      +- Project [dau#99L, revenue#100, app_id#101, event_date#102, __PARTITION_TIME__#130, __num_rows__#140]\n",
      "         +- Project [dau#99L, revenue#100, app_id#101, event_date#102, __PARTITION_TIME__#130, __num_rows__#140, __num_rows__#140]\n",
      "            +- Window [row_number() windowspecdefinition(__PARTITION_TIME__#130, __PARTITION_TIME__#130 ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS __num_rows__#140], [__PARTITION_TIME__#130], [__PARTITION_TIME__#130 ASC NULLS FIRST]\n",
      "               +- Project [dau#99L, revenue#100, app_id#101, event_date#102, __PARTITION_TIME__#130]\n",
      "                  +- Project [dau#99L, revenue#100, app_id#101, event_date#102, cast((cast(if (isnull(cast((unix_timestamp(event_date#102, yyyy-MM-dd HH:mm:ss, Some(UTC), false) * cast(1000 as bigint)) as bigint))) null else UDF(knownnotnull(cast((unix_timestamp(event_date#102, yyyy-MM-dd HH:mm:ss, Some(UTC), false) * cast(1000 as bigint)) as bigint))) as double) / cast(1000 as double)) as timestamp) AS __PARTITION_TIME__#130]\n",
      "                     +- LogicalRDD [dau#99L, revenue#100, app_id#101, event_date#102], false\n",
      "\n",
      "== Analyzed Logical Plan ==\n",
      "dau: bigint, revenue: double, app_id: string, event_date: timestamp, __PARTITION_TIME__: timestamp, __PARTITION_NUM__: int\n",
      "Project [dau#99L, revenue#100, app_id#101, event_date#102, __PARTITION_TIME__#130, __PARTITION_NUM__#147]\n",
      "+- RepartitionByExpression [__PARTITION_TIME__#130, __PARTITION_NUM__#147]\n",
      "   +- Project [dau#99L, revenue#100, app_id#101, event_date#102, __PARTITION_TIME__#130, __num_rows__#140, cast((cast((__num_rows__#140 - 1) as double) / cast(5000000 as double)) as int) AS __PARTITION_NUM__#147]\n",
      "      +- Project [dau#99L, revenue#100, app_id#101, event_date#102, __PARTITION_TIME__#130, __num_rows__#140]\n",
      "         +- Project [dau#99L, revenue#100, app_id#101, event_date#102, __PARTITION_TIME__#130, __num_rows__#140, __num_rows__#140]\n",
      "            +- Window [row_number() windowspecdefinition(__PARTITION_TIME__#130, __PARTITION_TIME__#130 ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS __num_rows__#140], [__PARTITION_TIME__#130], [__PARTITION_TIME__#130 ASC NULLS FIRST]\n",
      "               +- Project [dau#99L, revenue#100, app_id#101, event_date#102, __PARTITION_TIME__#130]\n",
      "                  +- Project [dau#99L, revenue#100, app_id#101, event_date#102, cast((cast(if (isnull(cast((unix_timestamp(event_date#102, yyyy-MM-dd HH:mm:ss, Some(UTC), false) * cast(1000 as bigint)) as bigint))) null else UDF(knownnotnull(cast((unix_timestamp(event_date#102, yyyy-MM-dd HH:mm:ss, Some(UTC), false) * cast(1000 as bigint)) as bigint))) as double) / cast(1000 as double)) as timestamp) AS __PARTITION_TIME__#130]\n",
      "                     +- LogicalRDD [dau#99L, revenue#100, app_id#101, event_date#102], false\n",
      "\n",
      "== Optimized Logical Plan ==\n",
      "RepartitionByExpression [__PARTITION_TIME__#130, __PARTITION_NUM__#147]\n",
      "+- Project [dau#99L, revenue#100, app_id#101, event_date#102, __PARTITION_TIME__#130, cast((cast((__num_rows__#140 - 1) as double) / 5000000.0) as int) AS __PARTITION_NUM__#147]\n",
      "   +- Window [row_number() windowspecdefinition(__PARTITION_TIME__#130, __PARTITION_TIME__#130 ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS __num_rows__#140], [__PARTITION_TIME__#130], [__PARTITION_TIME__#130 ASC NULLS FIRST]\n",
      "      +- Project [dau#99L, revenue#100, app_id#101, event_date#102, if (isnull((unix_timestamp(event_date#102, yyyy-MM-dd HH:mm:ss, Some(UTC), false) * 1000))) null else cast((cast(UDF(knownnotnull((unix_timestamp(event_date#102, yyyy-MM-dd HH:mm:ss, Some(UTC), false) * 1000))) as double) / 1000.0) as timestamp) AS __PARTITION_TIME__#130]\n",
      "         +- LogicalRDD [dau#99L, revenue#100, app_id#101, event_date#102], false\n",
      "\n",
      "== Physical Plan ==\n",
      "AdaptiveSparkPlan isFinalPlan=false\n",
      "+- Exchange hashpartitioning(__PARTITION_TIME__#130, __PARTITION_NUM__#147, 1000), REPARTITION_BY_COL, [id=#334]\n",
      "   +- Project [dau#99L, revenue#100, app_id#101, event_date#102, __PARTITION_TIME__#130, cast((cast((__num_rows__#140 - 1) as double) / 5000000.0) as int) AS __PARTITION_NUM__#147]\n",
      "      +- Window [row_number() windowspecdefinition(__PARTITION_TIME__#130, __PARTITION_TIME__#130 ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS __num_rows__#140], [__PARTITION_TIME__#130], [__PARTITION_TIME__#130 ASC NULLS FIRST]\n",
      "         +- Sort [__PARTITION_TIME__#130 ASC NULLS FIRST, __PARTITION_TIME__#130 ASC NULLS FIRST], false, 0\n",
      "            +- Exchange hashpartitioning(__PARTITION_TIME__#130, 1000), ENSURE_REQUIREMENTS, [id=#330]\n",
      "               +- Project [dau#99L, revenue#100, app_id#101, event_date#102, if (isnull((unix_timestamp(event_date#102, yyyy-MM-dd HH:mm:ss, Some(UTC), false) * 1000))) null else cast((cast(UDF(knownnotnull((unix_timestamp(event_date#102, yyyy-MM-dd HH:mm:ss, Some(UTC), false) * 1000))) as double) / 1000.0) as timestamp) AS __PARTITION_TIME__#130]\n",
      "                  +- Scan ExistingRDD[dau#99L,revenue#100,app_id#101,event_date#102]"
     ]
    }
   ],
   "source": [
    "%%spark\n",
    "\n",
    "from py4j.java_gateway import java_import\n",
    "from rovio_ingest import DRUID_SOURCE\n",
    "from rovio_ingest.extensions.dataframe_extension import ConfKeys, add_dataframe_druid_extension\n",
    "\n",
    "# fix df.explain on EMR 6\n",
    "java_import(spark._sc._jvm, \"org.apache.spark.sql.api.python.*\")\n",
    "\n",
    "add_dataframe_druid_extension()\n",
    "\n",
    "df_prepared = df.repartition_by_druid_segment_size('event_date', segment_granularity='DAY')\n",
    "df_prepared.explain(True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…",
      "application/vnd.jupyter.widget-view+json": {
       "version_major": 2,
       "version_minor": 0,
       "model_id": "388360a9b3ec432d9f472b8f3e89f458"
      }
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- dau: long (nullable = true)\n",
      " |-- revenue: double (nullable = true)\n",
      " |-- app_id: string (nullable = true)\n",
      " |-- event_date: timestamp (nullable = true)\n",
      " |-- __PARTITION_TIME__: timestamp (nullable = true)\n",
      " |-- __PARTITION_NUM__: integer (nullable = true)"
     ]
    }
   ],
   "source": [
    "%%spark\n",
    "df_prepared.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…",
      "application/vnd.jupyter.widget-view+json": {
       "version_major": 2,
       "version_minor": 0,
       "model_id": "5c0b71620e30443ab839935cf3567f0b"
      }
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-------+----------+-------------------+-------------------+-----------------+\n",
      "|dau|revenue|    app_id|         event_date| __PARTITION_TIME__|__PARTITION_NUM__|\n",
      "+---+-------+----------+-------------------+-------------------+-----------------+\n",
      "|  5|   30.0|testclient|2019-10-01 00:00:00|2019-10-01 00:00:00|                0|\n",
      "|  2|   15.0|testclient|2019-10-02 00:00:00|2019-10-02 00:00:00|                0|\n",
      "+---+-------+----------+-------------------+-------------------+-----------------+"
     ]
    }
   ],
   "source": [
    "%%spark\n",
    "df_prepared.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…",
      "application/vnd.jupyter.widget-view+json": {
       "version_major": 2,
       "version_minor": 0,
       "model_id": "1c4855ffa09c4fafaf2e8a4762aaa7d8"
      }
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "%%spark\n",
    "\n",
    "DATA_SOURCE_NAME = \"rovio_ingest_test_juho\"\n",
    "\n",
    "df_prepared \\\n",
    "    .write \\\n",
    "    .mode(\"overwrite\") \\\n",
    "    .format(DRUID_SOURCE) \\\n",
    "    .option(ConfKeys.DATA_SOURCE, DATA_SOURCE_NAME) \\\n",
    "    .option(ConfKeys.TIME_COLUMN, \"event_date\") \\\n",
    "    .option(ConfKeys.METADATA_DB_URI, get_param(\"druid/metadata_db/uri\")) \\\n",
    "    .option(ConfKeys.METADATA_DB_USERNAME, get_param(\"druid/metadata_db/username\")) \\\n",
    "    .option(ConfKeys.METADATA_DB_PASSWORD, get_param(\"druid/metadata_db/password\")) \\\n",
    "    .option(ConfKeys.DEEP_STORAGE_S3_BUCKET, get_param(\"druid/deep_storage/bucket\")) \\\n",
    "    .option(ConfKeys.DEEP_STORAGE_S3_BASE_KEY, \"druid/segments\") \\\n",
    "    .save()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "pycharm": {
     "is_executing": false,
     "name": "#%% md\n"
    }
   },
   "source": [
    "To list the written data you can run:\n",
    "\n",
    "    aws s3 --profile smoke ls --recursive \\\n",
    "      s3://{druid-deep-storage-bucket}/druid/segments/rovio_ingest_test_juho/\n",
    "\n",
    "To see something like:\n",
    "\n",
    "    2023-02-05 23:22:57       1058 druid/segments/rovio_ingest_test_juho/2019-10-01T00:00:00.000Z_2019-10-02T00:00:00.000Z/2023-02-05T21:22:52.965Z/0/index.zip\n",
    "    2023-02-05 23:22:57       1056 druid/segments/rovio_ingest_test_juho/2019-10-02T00:00:00.000Z_2019-10-03T00:00:00.000Z/2023-02-05T21:22:52.965Z/0/index.zip\n",
    "\n",
    "And run this in druid-sql (JDBC)\n",
    "\n",
    "    SELECT * FROM rovio_ingest_test_juho LIMIT 10;\n",
    "\n",
    "    __time\tapp_id\tdau\trevenue\n",
    "    2019-10-01 00:00:00\ttestclient\t5\t30\n",
    "    2019-10-02 00:00:00\ttestclient\t2\t15"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "%spark cleanup"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.7.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}