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",
    "## 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",
    "## 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_PROFILE=smoke\n",
    "        JAR_BUCKET=<REPLACE THIS>\n",
    "        aws s3 --profile $AWS_PROFILE cp \\\n",
    "          target/rovio-ingest-1.0.0_spark_3.0.1-SNAPSHOT.jar \\\n",
    "          s3://$JAR_BUCKET/tmp/juho/druid/jars/rovio-ingest-1.0.0_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.0_spark_3.0.1-SNAPSHOT.jar \\\n",
    "          s3://$JAR_BUCKET/tmp/juho/druid/jars/original-rovio-ingest-1.0.0_spark_3.0.1-SNAPSHOT.jar"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "pycharm": {
     "is_executing": false,
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "%load_ext sparkmagic.magics"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "pycharm": {
     "is_executing": false,
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "ENV = \"smoke\"\n",
    "PREFIX = \"tmp/juho/\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "pycharm": {
     "is_executing": false,
     "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": 4,
   "metadata": {
    "pycharm": {
     "is_executing": false,
     "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": 5,
   "metadata": {
    "pycharm": {
     "is_executing": false
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Starting Spark application\n"
     ]
    },
    {
     "data": {
      "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>19</td><td>application_1618230936742_0100</td><td>pyspark</td><td>idle</td><td><a target=\"_blank\" href=\"http://ip-10-0-10-245.cloud-dev.rovio.com:20888/proxy/application_1618230936742_0100/\">Link</a></td><td><a target=\"_blank\" href=\"http://ip-10-0-10-7.cloud-dev.rovio.com:8042/node/containerlogs/container_1618230936742_0100_01_000001/livy\">Link</a></td><td>✔</td></tr></table>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "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": [
      "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",
    "    \"spark.jars.repositories\": \"https://s01.oss.sonatype.org/content/repositories/snapshots\",\n",
    "    \"spark.jars.packages\": \"com.rovio.ingest:rovio-ingest:1.0.0_spark_3.0.1-SNAPSHOT\" \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.0_spark_3.0.1-SNAPSHOT.jar\"\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": 17,
   "metadata": {},
   "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"
    }
   ],
   "source": [
    "%%spark\n",
    "\n",
    "# This extension is provided by AWS EMR\n",
    "# spark.sparkContext.list_packages()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "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==0.0.1.dev14\n",
      "  Downloading https://files.pythonhosted.org/packages/c8/18/d2681cc3c550eebbb65aa14694bdab987a90682d090fca9008dd0776ba71/rovio_ingest-0.0.1.dev14-py3-none-any.whl\n",
      "Collecting pyspark<4.0.0,>=3.0.0 (from rovio-ingest==0.0.1.dev14)\n",
      "  Downloading https://files.pythonhosted.org/packages/45/b0/9d6860891ab14a39d4bddf80ba26ce51c2f9dc4805e5c6978ac0472c120a/pyspark-3.1.1.tar.gz (212.3MB)\n",
      "Collecting py4j==0.10.9 (from pyspark<4.0.0,>=3.0.0->rovio-ingest==0.0.1.dev14)\n",
      "  Downloading https://files.pythonhosted.org/packages/9e/b6/6a4fb90cd235dc8e265a6a2067f2a2c99f0d91787f06aca4bcf7c23f3f80/py4j-0.10.9-py2.py3-none-any.whl (198kB)\n",
      "Building wheels for collected packages: pyspark\n",
      "  Running setup.py bdist_wheel for pyspark: started\n",
      "  Running setup.py bdist_wheel for pyspark: finished with status 'done'\n",
      "  Stored in directory: /var/lib/livy/.cache/pip/wheels/0b/90/c0/01de724414ef122bd05f056541fb6a0ecf47c7ca655f8b3c0f\n",
      "Successfully built pyspark\n",
      "Installing collected packages: py4j, pyspark, rovio-ingest\n",
      "Successfully installed py4j-0.10.9 pyspark-3.1.1 rovio-ingest-0.0.1.dev14"
     ]
    }
   ],
   "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 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": 11,
   "metadata": {
    "pycharm": {
     "is_executing": false,
     "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"
    }
   ],
   "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": 12,
   "metadata": {
    "pycharm": {
     "is_executing": false
    }
   },
   "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": [
      "+---+-------+----------+-------------------+\n",
      "|dau|revenue|    app_id|         event_date|\n",
      "+---+-------+----------+-------------------+\n",
      "|  5|   30.0|testclient|2018-10-01 00:00:00|\n",
      "|  2|   15.0|testclient|2018-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(2018, 10, 1)],\n",
    "                            [2, 15.0, 'testclient', datetime(2018, 10, 2)]],\n",
    "                            schema)\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "pycharm": {
     "is_executing": false
    }
   },
   "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": [
      "== Parsed Logical Plan ==\n",
      "Project [dau#0L, revenue#1, app_id#2, event_date#3, __PARTITION_TIME__#30, __PARTITION_NUM__#44]\n",
      "+- RepartitionByExpression [__PARTITION_TIME__#30, __PARTITION_NUM__#44], 200\n",
      "   +- Project [dau#0L, revenue#1, app_id#2, event_date#3, __PARTITION_TIME__#30, __num_rows__#37, cast((cast((__num_rows__#37 - 1) as double) / cast(5000000 as double)) as int) AS __PARTITION_NUM__#44]\n",
      "      +- Project [dau#0L, revenue#1, app_id#2, event_date#3, __PARTITION_TIME__#30, __num_rows__#37]\n",
      "         +- Project [dau#0L, revenue#1, app_id#2, event_date#3, __PARTITION_TIME__#30, __num_rows__#37, __num_rows__#37]\n",
      "            +- Window [row_number() windowspecdefinition(__PARTITION_TIME__#30, __PARTITION_TIME__#30 ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS __num_rows__#37], [__PARTITION_TIME__#30], [__PARTITION_TIME__#30 ASC NULLS FIRST]\n",
      "               +- Project [dau#0L, revenue#1, app_id#2, event_date#3, __PARTITION_TIME__#30]\n",
      "                  +- Project [dau#0L, revenue#1, app_id#2, event_date#3, cast((cast(if (isnull(cast((unix_timestamp(event_date#3, yyyy-MM-dd HH:mm:ss, Some(UTC)) * cast(1000 as bigint)) as bigint))) null else UDF(knownnotnull(cast((unix_timestamp(event_date#3, yyyy-MM-dd HH:mm:ss, Some(UTC)) * cast(1000 as bigint)) as bigint))) as double) / cast(1000 as double)) as timestamp) AS __PARTITION_TIME__#30]\n",
      "                     +- LogicalRDD [dau#0L, revenue#1, app_id#2, event_date#3], 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#0L, revenue#1, app_id#2, event_date#3, __PARTITION_TIME__#30, __PARTITION_NUM__#44]\n",
      "+- RepartitionByExpression [__PARTITION_TIME__#30, __PARTITION_NUM__#44], 200\n",
      "   +- Project [dau#0L, revenue#1, app_id#2, event_date#3, __PARTITION_TIME__#30, __num_rows__#37, cast((cast((__num_rows__#37 - 1) as double) / cast(5000000 as double)) as int) AS __PARTITION_NUM__#44]\n",
      "      +- Project [dau#0L, revenue#1, app_id#2, event_date#3, __PARTITION_TIME__#30, __num_rows__#37]\n",
      "         +- Project [dau#0L, revenue#1, app_id#2, event_date#3, __PARTITION_TIME__#30, __num_rows__#37, __num_rows__#37]\n",
      "            +- Window [row_number() windowspecdefinition(__PARTITION_TIME__#30, __PARTITION_TIME__#30 ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS __num_rows__#37], [__PARTITION_TIME__#30], [__PARTITION_TIME__#30 ASC NULLS FIRST]\n",
      "               +- Project [dau#0L, revenue#1, app_id#2, event_date#3, __PARTITION_TIME__#30]\n",
      "                  +- Project [dau#0L, revenue#1, app_id#2, event_date#3, cast((cast(if (isnull(cast((unix_timestamp(event_date#3, yyyy-MM-dd HH:mm:ss, Some(UTC)) * cast(1000 as bigint)) as bigint))) null else UDF(knownnotnull(cast((unix_timestamp(event_date#3, yyyy-MM-dd HH:mm:ss, Some(UTC)) * cast(1000 as bigint)) as bigint))) as double) / cast(1000 as double)) as timestamp) AS __PARTITION_TIME__#30]\n",
      "                     +- LogicalRDD [dau#0L, revenue#1, app_id#2, event_date#3], false\n",
      "\n",
      "== Optimized Logical Plan ==\n",
      "RepartitionByExpression [__PARTITION_TIME__#30, __PARTITION_NUM__#44], 200\n",
      "+- Project [dau#0L, revenue#1, app_id#2, event_date#3, __PARTITION_TIME__#30, cast((cast((__num_rows__#37 - 1) as double) / 5000000.0) as int) AS __PARTITION_NUM__#44]\n",
      "   +- Window [row_number() windowspecdefinition(__PARTITION_TIME__#30, __PARTITION_TIME__#30 ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS __num_rows__#37], [__PARTITION_TIME__#30], [__PARTITION_TIME__#30 ASC NULLS FIRST]\n",
      "      +- Project [dau#0L, revenue#1, app_id#2, event_date#3, cast((cast(if (isnull((unix_timestamp(event_date#3, yyyy-MM-dd HH:mm:ss, Some(UTC)) * 1000))) null else UDF(knownnotnull((unix_timestamp(event_date#3, yyyy-MM-dd HH:mm:ss, Some(UTC)) * 1000))) as double) / 1000.0) as timestamp) AS __PARTITION_TIME__#30]\n",
      "         +- LogicalRDD [dau#0L, revenue#1, app_id#2, event_date#3], false\n",
      "\n",
      "== Physical Plan ==\n",
      "AdaptiveSparkPlan isFinalPlan=false\n",
      "+- Exchange hashpartitioning(__PARTITION_TIME__#30, __PARTITION_NUM__#44, 200), false, [id=#35]\n",
      "   +- Project [dau#0L, revenue#1, app_id#2, event_date#3, __PARTITION_TIME__#30, cast((cast((__num_rows__#37 - 1) as double) / 5000000.0) as int) AS __PARTITION_NUM__#44]\n",
      "      +- Window [row_number() windowspecdefinition(__PARTITION_TIME__#30, __PARTITION_TIME__#30 ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS __num_rows__#37], [__PARTITION_TIME__#30], [__PARTITION_TIME__#30 ASC NULLS FIRST]\n",
      "         +- Sort [__PARTITION_TIME__#30 ASC NULLS FIRST, __PARTITION_TIME__#30 ASC NULLS FIRST], false, 0\n",
      "            +- Exchange hashpartitioning(__PARTITION_TIME__#30, 1000), true, [id=#31]\n",
      "               +- Project [dau#0L, revenue#1, app_id#2, event_date#3, cast((cast(if (isnull((unix_timestamp(event_date#3, yyyy-MM-dd HH:mm:ss, Some(UTC)) * 1000))) null else UDF(knownnotnull((unix_timestamp(event_date#3, yyyy-MM-dd HH:mm:ss, Some(UTC)) * 1000))) as double) / 1000.0) as timestamp) AS __PARTITION_TIME__#30]\n",
      "                  +- Scan ExistingRDD[dau#0L,revenue#1,app_id#2,event_date#3]"
     ]
    }
   ],
   "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": 14,
   "metadata": {
    "pycharm": {
     "is_executing": false,
     "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": [
      "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": 15,
   "metadata": {
    "pycharm": {
     "is_executing": false,
     "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": [
      "+---+-------+----------+-------------------+-------------------+-----------------+\n",
      "|dau|revenue|    app_id|         event_date| __PARTITION_TIME__|__PARTITION_NUM__|\n",
      "+---+-------+----------+-------------------+-------------------+-----------------+\n",
      "|  2|   15.0|testclient|2018-10-02 00:00:00|2018-10-02 00:00:00|                0|\n",
      "|  5|   30.0|testclient|2018-10-01 00:00:00|2018-10-01 00:00:00|                0|\n",
      "+---+-------+----------+-------------------+-------------------+-----------------+"
     ]
    }
   ],
   "source": [
    "%%spark\n",
    "df_prepared.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "pycharm": {
     "is_executing": false,
     "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"
    }
   ],
   "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",
    "    2020-04-12 16:12:06        591 druid/segments/rovio_ingest_test_juho/2018-10-01T00:00:00.000Z_2018-10-02T00:00:00.000Z/2020-04-12T13:11:43.778Z/0/descriptor.json\n",
    "    2020-04-12 16:12:06       1055 druid/segments/rovio_ingest_test_juho/2018-10-01T00:00:00.000Z_2018-10-02T00:00:00.000Z/2020-04-12T13:11:43.778Z/0/index.zip\n",
    "    2020-04-12 16:12:06        591 druid/segments/rovio_ingest_test_juho/2018-10-02T00:00:00.000Z_2018-10-03T00:00:00.000Z/2020-04-12T13:11:43.778Z/0/descriptor.json\n",
    "    2020-04-12 16:12:06       1052 druid/segments/rovio_ingest_test_juho/2018-10-02T00:00:00.000Z_2018-10-03T00:00:00.000Z/2020-04-12T13:11:43.778Z/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",
    "    2018-10-01 00:00:00\ttestclient\t5\t30\n",
    "    2018-10-02 00:00:00\ttestclient\t2\t15"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "pycharm": {
     "is_executing": false,
     "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
}