smoketest.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Requirement already satisfied: kafka-python in /home/merlijn/.local/lib/python3.9/site-packages (2.0.2)\n",
      "Requirement already satisfied: flask in /home/merlijn/.local/lib/python3.9/site-packages (2.0.2)\n",
      "Requirement already satisfied: Werkzeug>=2.0 in /home/merlijn/.local/lib/python3.9/site-packages (from flask) (2.0.2)\n",
      "Requirement already satisfied: itsdangerous>=2.0 in /home/merlijn/.local/lib/python3.9/site-packages (from flask) (2.0.1)\n",
      "Requirement already satisfied: Jinja2>=3.0 in /home/merlijn/.local/lib/python3.9/site-packages (from flask) (3.0.1)\n",
      "Requirement already satisfied: click>=7.1.2 in /usr/lib/python3/dist-packages (from flask) (7.1.2)\n",
      "Requirement already satisfied: MarkupSafe>=2.0 in /home/merlijn/.local/lib/python3.9/site-packages (from Jinja2>=3.0->flask) (2.0.1)\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    "# Install the required Python 3 dependencies\n",
    "python3 -m pip install kafka-python flask  # type: ignore"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "RecordMetadata(topic='smoke-test', partition=0, topic_partition=TopicPartition(topic='smoke-test', partition=0), offset=2, timestamp=1637862851980, log_start_offset=0, checksum=None, serialized_key_size=-1, serialized_value_size=3, serialized_header_size=-1)"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from kafka import KafkaProducer\n",
    "\n",
    "topic = \"smoke-test\"\n",
    "producer = KafkaProducer(bootstrap_servers=['localhost:9092'])\n",
    "# Note: The data you send must be binary\n",
    "producer.send(topic, b\"Hello World!\").get(timeout=30)\n",
    "# Note: The data you send must be binary\n",
    "producer.send(topic, b\"Foo\").get(timeout=30)\n",
    "producer.send(topic, b\"Bar\").get(timeout=30)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING: An illegal reflective access operation has occurred\n",
      "WARNING: Illegal reflective access by org.apache.spark.unsafe.Platform (file:/usr/local/spark-3.2.0-bin-hadoop3.2/jars/spark-unsafe_2.12-3.2.0.jar) to constructor java.nio.DirectByteBuffer(long,int)\n",
      "WARNING: Please consider reporting this to the maintainers of org.apache.spark.unsafe.Platform\n",
      "WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations\n",
      "WARNING: All illegal access operations will be denied in a future release\n",
      "Ivy Default Cache set to: /home/jovyan/.ivy2/cache\n",
      "The jars for the packages stored in: /home/jovyan/.ivy2/jars\n",
      "org.apache.spark#spark-sql-kafka-0-10_2.12 added as a dependency\n",
      ":: resolving dependencies :: org.apache.spark#spark-submit-parent-38f23ba4-2a1d-4ed8-ad61-18acc10e556b;1.0\n",
      "\tconfs: [default]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      ":: loading settings :: url = jar:file:/usr/local/spark-3.2.0-bin-hadoop3.2/jars/ivy-2.5.0.jar!/org/apache/ivy/core/settings/ivysettings.xml\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\tfound org.apache.spark#spark-sql-kafka-0-10_2.12;3.2.0 in central\n",
      "\tfound org.apache.spark#spark-token-provider-kafka-0-10_2.12;3.2.0 in central\n",
      "\tfound org.apache.kafka#kafka-clients;2.8.0 in central\n",
      "\tfound org.lz4#lz4-java;1.7.1 in central\n",
      "\tfound org.xerial.snappy#snappy-java;1.1.8.4 in central\n",
      "\tfound org.slf4j#slf4j-api;1.7.30 in central\n",
      "\tfound org.apache.hadoop#hadoop-client-runtime;3.3.1 in central\n",
      "\tfound org.spark-project.spark#unused;1.0.0 in central\n",
      "\tfound org.apache.hadoop#hadoop-client-api;3.3.1 in central\n",
      "\tfound org.apache.htrace#htrace-core4;4.1.0-incubating in central\n",
      "\tfound commons-logging#commons-logging;1.1.3 in central\n",
      "\tfound com.google.code.findbugs#jsr305;3.0.0 in central\n",
      "\tfound org.apache.commons#commons-pool2;2.6.2 in central\n",
      "downloading https://repo1.maven.org/maven2/org/apache/spark/spark-sql-kafka-0-10_2.12/3.2.0/spark-sql-kafka-0-10_2.12-3.2.0.jar ...\n",
      "\t[SUCCESSFUL ] org.apache.spark#spark-sql-kafka-0-10_2.12;3.2.0!spark-sql-kafka-0-10_2.12.jar (76ms)\n",
      "downloading https://repo1.maven.org/maven2/org/apache/spark/spark-token-provider-kafka-0-10_2.12/3.2.0/spark-token-provider-kafka-0-10_2.12-3.2.0.jar ...\n",
      "\t[SUCCESSFUL ] org.apache.spark#spark-token-provider-kafka-0-10_2.12;3.2.0!spark-token-provider-kafka-0-10_2.12.jar (24ms)\n",
      "downloading https://repo1.maven.org/maven2/org/apache/kafka/kafka-clients/2.8.0/kafka-clients-2.8.0.jar ...\n",
      "\t[SUCCESSFUL ] org.apache.kafka#kafka-clients;2.8.0!kafka-clients.jar (668ms)\n",
      "downloading https://repo1.maven.org/maven2/com/google/code/findbugs/jsr305/3.0.0/jsr305-3.0.0.jar ...\n",
      "\t[SUCCESSFUL ] com.google.code.findbugs#jsr305;3.0.0!jsr305.jar (34ms)\n",
      "downloading https://repo1.maven.org/maven2/org/apache/commons/commons-pool2/2.6.2/commons-pool2-2.6.2.jar ...\n",
      "\t[SUCCESSFUL ] org.apache.commons#commons-pool2;2.6.2!commons-pool2.jar (35ms)\n",
      "downloading https://repo1.maven.org/maven2/org/spark-project/spark/unused/1.0.0/unused-1.0.0.jar ...\n",
      "\t[SUCCESSFUL ] org.spark-project.spark#unused;1.0.0!unused.jar (17ms)\n",
      "downloading https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-client-runtime/3.3.1/hadoop-client-runtime-3.3.1.jar ...\n",
      "\t[SUCCESSFUL ] org.apache.hadoop#hadoop-client-runtime;3.3.1!hadoop-client-runtime.jar (4222ms)\n",
      "downloading https://repo1.maven.org/maven2/org/lz4/lz4-java/1.7.1/lz4-java-1.7.1.jar ...\n",
      "\t[SUCCESSFUL ] org.lz4#lz4-java;1.7.1!lz4-java.jar (100ms)\n",
      "downloading https://repo1.maven.org/maven2/org/xerial/snappy/snappy-java/1.1.8.4/snappy-java-1.1.8.4.jar ...\n",
      "\t[SUCCESSFUL ] org.xerial.snappy#snappy-java;1.1.8.4!snappy-java.jar(bundle) (271ms)\n",
      "downloading https://repo1.maven.org/maven2/org/slf4j/slf4j-api/1.7.30/slf4j-api-1.7.30.jar ...\n",
      "\t[SUCCESSFUL ] org.slf4j#slf4j-api;1.7.30!slf4j-api.jar (23ms)\n",
      "downloading https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-client-api/3.3.1/hadoop-client-api-3.3.1.jar ...\n",
      "\t[SUCCESSFUL ] org.apache.hadoop#hadoop-client-api;3.3.1!hadoop-client-api.jar (2574ms)\n",
      "downloading https://repo1.maven.org/maven2/org/apache/htrace/htrace-core4/4.1.0-incubating/htrace-core4-4.1.0-incubating.jar ...\n",
      "\t[SUCCESSFUL ] org.apache.htrace#htrace-core4;4.1.0-incubating!htrace-core4.jar (229ms)\n",
      "downloading https://repo1.maven.org/maven2/commons-logging/commons-logging/1.1.3/commons-logging-1.1.3.jar ...\n",
      "\t[SUCCESSFUL ] commons-logging#commons-logging;1.1.3!commons-logging.jar (26ms)\n",
      ":: resolution report :: resolve 10748ms :: artifacts dl 8310ms\n",
      "\t:: modules in use:\n",
      "\tcom.google.code.findbugs#jsr305;3.0.0 from central in [default]\n",
      "\tcommons-logging#commons-logging;1.1.3 from central in [default]\n",
      "\torg.apache.commons#commons-pool2;2.6.2 from central in [default]\n",
      "\torg.apache.hadoop#hadoop-client-api;3.3.1 from central in [default]\n",
      "\torg.apache.hadoop#hadoop-client-runtime;3.3.1 from central in [default]\n",
      "\torg.apache.htrace#htrace-core4;4.1.0-incubating from central in [default]\n",
      "\torg.apache.kafka#kafka-clients;2.8.0 from central in [default]\n",
      "\torg.apache.spark#spark-sql-kafka-0-10_2.12;3.2.0 from central in [default]\n",
      "\torg.apache.spark#spark-token-provider-kafka-0-10_2.12;3.2.0 from central in [default]\n",
      "\torg.lz4#lz4-java;1.7.1 from central in [default]\n",
      "\torg.slf4j#slf4j-api;1.7.30 from central in [default]\n",
      "\torg.spark-project.spark#unused;1.0.0 from central in [default]\n",
      "\torg.xerial.snappy#snappy-java;1.1.8.4 from central in [default]\n",
      "\t---------------------------------------------------------------------\n",
      "\t|                  |            modules            ||   artifacts   |\n",
      "\t|       conf       | number| search|dwnlded|evicted|| number|dwnlded|\n",
      "\t---------------------------------------------------------------------\n",
      "\t|      default     |   13  |   13  |   13  |   0   ||   13  |   13  |\n",
      "\t---------------------------------------------------------------------\n",
      ":: retrieving :: org.apache.spark#spark-submit-parent-38f23ba4-2a1d-4ed8-ad61-18acc10e556b\n",
      "\tconfs: [default]\n",
      "\t13 artifacts copied, 0 already retrieved (59181kB/43ms)\n",
      "21/11/25 17:54:32 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
      "Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties\n",
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n",
      "21/11/25 17:54:35 WARN ResolveWriteToStream: Temporary checkpoint location created which is deleted normally when the query didn't fail: /tmp/temporary-876e4f53-f33c-423c-9402-5df055cdbe0e. If it's required to delete it under any circumstances, please set spark.sql.streaming.forceDeleteTempCheckpointLocation to true. Important to know deleting temp checkpoint folder is best effort.\n",
      "21/11/25 17:54:35 WARN ResolveWriteToStream: spark.sql.adaptive.enabled is not supported in streaming DataFrames/Datasets and will be disabled.\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "{'message': 'Waiting for data to arrive',\n",
       " 'isDataAvailable': False,\n",
       " 'isTriggerActive': False}"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>key</th>\n",
       "      <th>value</th>\n",
       "      <th>topic</th>\n",
       "      <th>partition</th>\n",
       "      <th>offset</th>\n",
       "      <th>timestamp</th>\n",
       "      <th>timestampType</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>None</td>\n",
       "      <td>[72, 101, 108, 108, 111, 32, 87, 111, 114, 108...</td>\n",
       "      <td>smoke-test</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>2021-11-25 17:54:11.948</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>None</td>\n",
       "      <td>[70, 111, 111]</td>\n",
       "      <td>smoke-test</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>2021-11-25 17:54:11.977</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>None</td>\n",
       "      <td>[66, 97, 114]</td>\n",
       "      <td>smoke-test</td>\n",
       "      <td>0</td>\n",
       "      <td>2</td>\n",
       "      <td>2021-11-25 17:54:11.980</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "    key                                              value       topic  \\\n",
       "0  None  [72, 101, 108, 108, 111, 32, 87, 111, 114, 108...  smoke-test   \n",
       "1  None                                     [70, 111, 111]  smoke-test   \n",
       "2  None                                      [66, 97, 114]  smoke-test   \n",
       "\n",
       "   partition  offset               timestamp  timestampType  \n",
       "0          0       0 2021-11-25 17:54:11.948              0  \n",
       "1          0       1 2021-11-25 17:54:11.977              0  \n",
       "2          0       2 2021-11-25 17:54:11.980              0  "
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "from IPython.display import display, clear_output\n",
    "from time import sleep\n",
    "\n",
    "import os\n",
    "os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.2.0 pyspark-shell'\n",
    "os.environ['SPARK_OPTS'] = \"--driver-java-options=-Xms1024M --driver-java-options=-Xmx4096M --driver-java-options=-Dlog4j.logLevel=warn\"\n",
    "\n",
    "\n",
    "import pyspark \n",
    "from pyspark import SparkContext\n",
    "from pyspark.sql.session import SparkSession\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.sql.functions import *\n",
    "\n",
    "spark = SparkSession.builder.master('local[2]').getOrCreate()\n",
    "sc = spark.sparkContext\n",
    "\n",
    "df = (\n",
    "    spark.readStream.format(\"kafka\")\n",
    "    .option(\"kafka.bootstrap.servers\",\"localhost:9092\")\n",
    "    .option(\"subscribe\", \"smoke-test\")\n",
    "    .option(\"startingOffsets\", \"earliest\")\n",
    "    .load()\n",
    ")\n",
    "\n",
    "smoke_test_query = (\n",
    "    df.writeStream                  # Create an output stream for the query\n",
    "    .outputMode(\"append\")           # Add each msg to end of \"streaming table\"\n",
    "    .format(\"memory\")               # Write output stream to Spark table\n",
    "    .queryName(\"test_query\") \n",
    "    .start()\n",
    ")\n",
    "\n",
    "# Wait until query runs for a bit\n",
    "sleep(10)\n",
    "\n",
    "# Display status of query\n",
    "display(smoke_test_query.status)\n",
    "\n",
    "# Show result in Jupyter Notebook\n",
    "display(\n",
    "    spark.table(\"test_query\").toPandas()\n",
    ")\n",
    "\n",
    "# Stop query\n",
    "smoke_test_query.stop()\n"
   ]
  }
 ],
 "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.9.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}