spark/end-to-end-spark-structured-streaming-kafka.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "4e652d46",
   "metadata": {},
   "source": [
    "# End to End Spark Structured Streaming for Kafka Topcis\n",
    "\n",
    "In the previous [notebook](spark-streaming.ipynb) we saw how to consume kafka events in Spark's structured streaming, in this Notebook we will look at how to create Kafka topic events and consume them into MinIO end to end with Spark's structured streaming without using the Kafka produceres or connectors"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b6ac98e9",
   "metadata": {},
   "source": [
    "### Prerequisites\n",
    "\n",
    "Before we get started we need the following ready\n",
    "\n",
    "1. [Kafka](../kafka/setup-kafka.ipynb)\n",
    "2. [Kafka Schema Registry](../kafka/kafka-schema-registry-minio.ipynb)\n",
    "3. [Spark Operator](setup-spark-operator.ipynb)\n",
    "4. MinIO cluster\n",
    "\n",
    "After we have the above prerequisites ready we will do the following\n",
    "\n",
    "* Modify Kafka Topic Partion\n",
    "* Spark Structured Streaming Consumer\n",
    "* Spark Structured Streaming Producer\n",
    "\n",
    "\n",
    "\n",
    "### Modify Kafka Topic Partition\n",
    "\n",
    "In order for us to gain full parallelization capabilities from Spark we will need to modify the `nyc-avro-topic` kafka topic partitions to `10` so that Spark Structured streaming 10 workers can pull data from Kafka simultaneously as show below."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "45cfe382",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting sample-code/spark-job/kafka-nyc-avro-topic.yaml\n"
     ]
    }
   ],
   "source": [
    "%%writefile sample-code/spark-job/kafka-nyc-avro-topic.yaml\n",
    "apiVersion: kafka.strimzi.io/v1beta2\n",
    "kind: KafkaTopic\n",
    "metadata:\n",
    "  name: nyc-avro-topic\n",
    "  namespace: kafka\n",
    "  labels:\n",
    "    strimzi.io/cluster: my-kafka-cluster\n",
    "spec:\n",
    "  partitions: 10\n",
    "  replicas: 3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5c284d2b",
   "metadata": {},
   "outputs": [],
   "source": [
    "!kubectl apply -f sample-code/spark-job/kafka-nyc-avro-topic.yaml"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d3707c60",
   "metadata": {},
   "source": [
    "**Note**: Before you apply the above change it is highly recommended to delete the `nyc-avro-topic` if it already exists based on the run from previous notebooks"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b2104510",
   "metadata": {},
   "source": [
    "### Spark Structured Streaming Consumer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "5500a608",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting sample-code/src/main-streaming-spark-consumer.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile sample-code/src/main-streaming-spark-consumer.py\n",
    "import os\n",
    "from pyspark import SparkConf\n",
    "from pyspark import SparkContext\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.avro.functions import from_avro\n",
    "from pyspark.sql.types import StructType, StructField, LongType, DoubleType, StringType\n",
    "import json\n",
    "\n",
    "conf = (\n",
    "    SparkConf()\n",
    "    .set(\"spark.sql.streaming.checkpointFileManagerClass\", \"io.minio.spark.checkpoint.S3BasedCheckpointFileManager\")\n",
    ")\n",
    "\n",
    "spark = SparkSession.builder.config(conf=conf).getOrCreate()\n",
    "\n",
    "\n",
    "def load_config(spark_context: SparkContext):\n",
    "    spark_context._jsc.hadoopConfiguration().set(\"fs.s3a.access.key\", \"openlakeuser\")\n",
    "    spark_context._jsc.hadoopConfiguration().set(\"fs.defaultFS\", \"s3://warehouse-v\")\n",
    "    spark_context._jsc.hadoopConfiguration().set(\"fs.s3a.secret.key\", \"openlakeuser\")\n",
    "    spark_context._jsc.hadoopConfiguration().set(\"fs.s3a.endpoint\", \"opl.ic.min.dev\")\n",
    "    spark_context._jsc.hadoopConfiguration().set(\"fs.s3a.connection.ssl.enabled\", \"true\")\n",
    "    spark_context._jsc.hadoopConfiguration().set(\"fs.s3a.path.style.access\", \"true\")\n",
    "    spark_context._jsc.hadoopConfiguration().set(\"fs.s3a.attempts.maximum\", \"1\")\n",
    "    spark_context._jsc.hadoopConfiguration().set(\"fs.s3a.connection.establish.timeout\", \"5000\")\n",
    "    spark_context._jsc.hadoopConfiguration().set(\"fs.s3a.connection.timeout\", \"10000\")\n",
    "\n",
    "\n",
    "load_config(spark.sparkContext)\n",
    "\n",
    "schema = StructType([\n",
    "    StructField('VendorID', LongType(), True),\n",
    "    StructField('tpep_pickup_datetime', StringType(), True),\n",
    "    StructField('tpep_dropoff_datetime', StringType(), True),\n",
    "    StructField('passenger_count', DoubleType(), True),\n",
    "    StructField('trip_distance', DoubleType(), True),\n",
    "    StructField('RatecodeID', DoubleType(), True),\n",
    "    StructField('store_and_fwd_flag', StringType(), True),\n",
    "    StructField('PULocationID', LongType(), True),\n",
    "    StructField('DOLocationID', LongType(), True),\n",
    "    StructField('payment_type', LongType(), True),\n",
    "    StructField('fare_amount', DoubleType(), True),\n",
    "    StructField('extra', DoubleType(), True),\n",
    "    StructField('mta_tax', DoubleType(), True),\n",
    "    StructField('tip_amount', DoubleType(), True),\n",
    "    StructField('tolls_amount', DoubleType(), True),\n",
    "    StructField('improvement_surcharge', DoubleType(), True),\n",
    "    StructField('total_amount', DoubleType(), True)])\n",
    "\n",
    "value_schema_dict = {\n",
    "    \"type\": \"record\",\n",
    "    \"name\": \"nyc_avro_test\",\n",
    "    \"fields\": [\n",
    "        {\n",
    "            \"name\": \"VendorID\",\n",
    "            \"type\": \"long\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"tpep_pickup_datetime\",\n",
    "            \"type\": \"string\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"tpep_dropoff_datetime\",\n",
    "            \"type\": \"string\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"passenger_count\",\n",
    "            \"type\": \"double\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"trip_distance\",\n",
    "            \"type\": \"double\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"RatecodeID\",\n",
    "            \"type\": \"double\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"store_and_fwd_flag\",\n",
    "            \"type\": \"string\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"PULocationID\",\n",
    "            \"type\": \"long\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"DOLocationID\",\n",
    "            \"type\": \"long\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"payment_type\",\n",
    "            \"type\": \"long\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"fare_amount\",\n",
    "            \"type\": \"double\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"extra\",\n",
    "            \"type\": \"double\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"mta_tax\",\n",
    "            \"type\": \"double\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"tip_amount\",\n",
    "            \"type\": \"double\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"tolls_amount\",\n",
    "            \"type\": \"double\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"improvement_surcharge\",\n",
    "            \"type\": \"double\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"total_amount\",\n",
    "            \"type\": \"double\"\n",
    "        },\n",
    "    ]\n",
    "}\n",
    "\n",
    "stream_df = spark.readStream.format(\"kafka\") \\\n",
    "    .option(\"kafka.bootstrap.servers\", \n",
    "            os.getenv(\"KAFKA_BOOTSTRAM_SERVER\",\"my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9092\")) \\\n",
    "    .option(\"subscribe\", \"nyc-avro-topic\") \\\n",
    "    .option(\"startingOffsets\", \"earliest\") \\\n",
    "    .option(\"failOnDataLoss\", \"false\") \\\n",
    "    .option(\"minPartitions\", \"10\") \\\n",
    "    .option(\"mode\", \"PERMISSIVE\") \\\n",
    "    .option(\"truncate\", False) \\\n",
    "    .option(\"newRows\", 100000) \\\n",
    "    .load()\n",
    "\n",
    "stream_df.printSchema()\n",
    "\n",
    "taxi_df = stream_df.select(from_avro(\"value\", json.dumps(value_schema_dict)).alias(\"data\")).select(\"data.*\")\n",
    "\n",
    "taxi_df.writeStream \\\n",
    "    .format(\"parquet\") \\\n",
    "    .outputMode(\"append\") \\\n",
    "    .trigger(processingTime='1 second') \\\n",
    "    .option(\"path\", \"s3a://warehouse-v/k8/spark-stream/\") \\\n",
    "    .option(\"checkpointLocation\", \"s3a://warehouse-v/k8/checkpoint\") \\\n",
    "    .start() \\\n",
    "    .awaitTermination()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3d93ba8f",
   "metadata": {},
   "source": [
    "As you can see from the above code there is a slight change in the code\n",
    "\n",
    "```python\n",
    "taxi_df = stream_df.select(from_avro(\"value\", json.dumps(value_schema_dict)).alias(\"data\")).select(\"data.*\")\n",
    "\n",
    "taxi_df.writeStream \\\n",
    "    .format(\"parquet\") \\\n",
    "    .outputMode(\"append\") \\\n",
    "    .trigger(processingTime='1 second') \\\n",
    "    .option(\"path\", \"s3a://warehouse-v/k8/spark-stream/\") \\\n",
    "    .option(\"checkpointLocation\", \"s3a://warehouse-v/k8/checkpoint\") \\\n",
    "    .start() \\\n",
    "    .awaitTermination()\n",
    "```\n",
    "\n",
    "The Avro is based on the Spark implementation so no need for Confluence dependency and skip the first 6 bits like we did earlier. We also added a `1 second` delay before polling for kafka events each time "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "d4b697a5",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting sample-code/Dockerfile\n"
     ]
    }
   ],
   "source": [
    "%%writefile sample-code/Dockerfile\n",
    "FROM openlake/spark-py:3.3.2\n",
    "\n",
    "USER root\n",
    "\n",
    "WORKDIR /app\n",
    "\n",
    "RUN pip3 install pyspark==3.3.2\n",
    "\n",
    "# Add avro dependency\n",
    "ADD https://repo1.maven.org/maven2/org/apache/commons/commons-pool2/2.11.1/commons-pool2-2.11.1.jar $SPARK_HOME/jars\n",
    "ADD https://repo1.maven.org/maven2/org/apache/spark/spark-avro_2.12/3.3.2/spark-avro_2.12-3.3.2.jar $SPARK_HOME/jars\n",
    "\n",
    "COPY src/*.py ."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b7447781",
   "metadata": {},
   "source": [
    "Build the Docker image with the above code or use `openlake/sparkjob-demo:3.3.2`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "d6f7e031",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Writing sample-code/spark-job/sparkjob-streaming-consumer.yaml\n"
     ]
    }
   ],
   "source": [
    "%%writefile sample-code/spark-job/sparkjob-streaming-consumer.yaml\n",
    "apiVersion: \"sparkoperator.k8s.io/v1beta2\"\n",
    "kind: SparkApplication\n",
    "metadata:\n",
    "  name: spark-stream-optimized\n",
    "  namespace: spark-operator\n",
    "spec:\n",
    "  type: Python\n",
    "  pythonVersion: \"3\"\n",
    "  mode: cluster\n",
    "  image: \"openlake/sparkjob-demo:3.3.2\"\n",
    "  imagePullPolicy: Always\n",
    "  mainApplicationFile: local:///app/main-streaming-spark-consumer.py\n",
    "  sparkVersion: \"3.3.2\"\n",
    "  restartPolicy:\n",
    "    type: OnFailure\n",
    "    onFailureRetries: 1\n",
    "    onFailureRetryInterval: 10\n",
    "    onSubmissionFailureRetries: 5\n",
    "    onSubmissionFailureRetryInterval: 20\n",
    "  driver:\n",
    "    cores: 3\n",
    "    memory: \"2048m\"\n",
    "    labels:\n",
    "      version: 3.3.2\n",
    "    serviceAccount: my-release-spark\n",
    "    env:\n",
    "      - name: AWS_REGION\n",
    "        value: us-east-1\n",
    "      - name: AWS_ACCESS_KEY_ID\n",
    "        value: openlakeuser\n",
    "      - name: AWS_SECRET_ACCESS_KEY\n",
    "        value: openlakeuser\n",
    "  executor:\n",
    "    cores: 1\n",
    "    instances: 10\n",
    "    memory: \"1024m\"\n",
    "    labels:\n",
    "      version: 3.3.2\n",
    "    env:\n",
    "      - name: AWS_REGION\n",
    "        value: us-east-1\n",
    "      - name: AWS_ACCESS_KEY_ID\n",
    "        value: openlakeuser\n",
    "      - name: AWS_SECRET_ACCESS_KEY\n",
    "        value: openlakeuser\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1d0e57e7",
   "metadata": {},
   "source": [
    "### Spark Structured Streaming Producer\n",
    "\n",
    "Now that we have the kafka topic configured correctly, let create a Kafka Producer using Spark Structured Streaming as shown below"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "a3e4c270",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Writing sample-code/src/spark-streaming-kafka-producer.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile sample-code/src/spark-streaming-kafka-producer.py\n",
    "import json\n",
    "from io import BytesIO\n",
    "import os\n",
    "import avro.schema\n",
    "from pyspark import SparkContext\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.avro.functions import to_avro\n",
    "from pyspark.sql.functions import struct\n",
    "from pyspark.sql.types import StructType, StructField, LongType, DoubleType, StringType\n",
    "\n",
    "spark = SparkSession.builder.getOrCreate()\n",
    "\n",
    "\n",
    "def serialize_avro(data, schema):\n",
    "    writer = avro.io.DatumWriter(schema)\n",
    "    bytes_writer = BytesIO()\n",
    "    encoder = avro.io.BinaryEncoder(bytes_writer)\n",
    "    writer.write(data, encoder)\n",
    "    return bytes_writer.getvalue()\n",
    "\n",
    "\n",
    "def load_config(spark_context: SparkContext):\n",
    "    spark_context._jsc.hadoopConfiguration().set(\"fs.s3a.access.key\", \"openlakeuser\")\n",
    "    spark_context._jsc.hadoopConfiguration().set(\"fs.defaultFS\", \"s3://warehouse-v\")\n",
    "    spark_context._jsc.hadoopConfiguration().set(\"fs.s3a.secret.key\", \"openlakeuser\")\n",
    "    spark_context._jsc.hadoopConfiguration().set(\"fs.s3a.endpoint\", \"opl.ic.min.dev\")\n",
    "    spark_context._jsc.hadoopConfiguration().set(\"fs.s3a.connection.ssl.enabled\", \"true\")\n",
    "    spark_context._jsc.hadoopConfiguration().set(\"fs.s3a.path.style.access\", \"true\")\n",
    "    spark_context._jsc.hadoopConfiguration().set(\"fs.s3a.attempts.maximum\", \"1\")\n",
    "    spark_context._jsc.hadoopConfiguration().set(\"fs.s3a.connection.establish.timeout\", \"5000\")\n",
    "    spark_context._jsc.hadoopConfiguration().set(\"fs.s3a.connection.timeout\", \"10000\")\n",
    "\n",
    "\n",
    "load_config(spark.sparkContext)\n",
    "\n",
    "schema = StructType([\n",
    "    StructField('VendorID', LongType(), True),\n",
    "    StructField('tpep_pickup_datetime', StringType(), True),\n",
    "    StructField('tpep_dropoff_datetime', StringType(), True),\n",
    "    StructField('passenger_count', DoubleType(), True),\n",
    "    StructField('trip_distance', DoubleType(), True),\n",
    "    StructField('RatecodeID', DoubleType(), True),\n",
    "    StructField('store_and_fwd_flag', StringType(), True),\n",
    "    StructField('PULocationID', LongType(), True),\n",
    "    StructField('DOLocationID', LongType(), True),\n",
    "    StructField('payment_type', LongType(), True),\n",
    "    StructField('fare_amount', DoubleType(), True),\n",
    "    StructField('extra', DoubleType(), True),\n",
    "    StructField('mta_tax', DoubleType(), True),\n",
    "    StructField('tip_amount', DoubleType(), True),\n",
    "    StructField('tolls_amount', DoubleType(), True),\n",
    "    StructField('improvement_surcharge', DoubleType(), True),\n",
    "    StructField('total_amount', DoubleType(), True)])\n",
    "\n",
    "value_schema_dict = {\n",
    "    \"type\": \"record\",\n",
    "    \"name\": \"nyc_avro_test\",\n",
    "    \"fields\": [\n",
    "        {\n",
    "            \"name\": \"VendorID\",\n",
    "            \"type\": \"long\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"tpep_pickup_datetime\",\n",
    "            \"type\": \"string\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"tpep_dropoff_datetime\",\n",
    "            \"type\": \"string\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"passenger_count\",\n",
    "            \"type\": \"double\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"trip_distance\",\n",
    "            \"type\": \"double\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"RatecodeID\",\n",
    "            \"type\": \"double\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"store_and_fwd_flag\",\n",
    "            \"type\": \"string\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"PULocationID\",\n",
    "            \"type\": \"long\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"DOLocationID\",\n",
    "            \"type\": \"long\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"payment_type\",\n",
    "            \"type\": \"long\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"fare_amount\",\n",
    "            \"type\": \"double\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"extra\",\n",
    "            \"type\": \"double\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"mta_tax\",\n",
    "            \"type\": \"double\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"tip_amount\",\n",
    "            \"type\": \"double\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"tolls_amount\",\n",
    "            \"type\": \"double\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"improvement_surcharge\",\n",
    "            \"type\": \"double\"\n",
    "        },\n",
    "        {\n",
    "            \"name\": \"total_amount\",\n",
    "            \"type\": \"double\"\n",
    "        },\n",
    "    ]\n",
    "}\n",
    "\n",
    "value_schema_str = json.dumps(value_schema_dict)\n",
    "\n",
    "df = spark.read.option(\"header\", \"true\").schema(schema).csv(\n",
    "    os.getenv(\"INPUT_PATH\", \"s3a://openlake/spark/sample-data/taxi-data.csv\"))\n",
    "df = df.select(to_avro(struct([df[x] for x in df.columns]), value_schema_str).alias(\"value\"))\n",
    "\n",
    "df.write \\\n",
    "    .format(\"kafka\") \\\n",
    "    .option(\"kafka.bootstrap.servers\", \n",
    "            os.getenv(\"KAFKA_BOOTSTRAM_SERVER\", \"my-kafka-cluster-kafka-bootstrap.kafka.svc.cluster.local:9092\")) \\\n",
    "    .option(\"flushInterval\", \"100ms\") \\\n",
    "    .option(\"key.serializer\", \"org.apache.kafka.common.serialization.StringSerializer\") \\\n",
    "    .option(\"value.serializer\", \"io.confluent.kafka.serializers.KafkaAvroSerializer\") \\\n",
    "    .option(\"schema.registry.url\",\n",
    "            os.getenv(\"KAFKA_SCHEMA_REGISTRY\", \"http://kafka-schema-registry-cp-schema-registry.kafka.svc.cluster.local:8081\")) \\\n",
    "    .option(\"topic\", \"nyc-avro-topic\") \\\n",
    "    .save()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ab029013",
   "metadata": {},
   "source": [
    "In the above code we read the `taxi-data.csv` data from MinIO bucket in this block\n",
    "\n",
    "```python\n",
    "df = spark.read.option(\"header\", \"true\").schema(schema).csv(\n",
    "    os.getenv(\"INPUT_PATH\", \"s3a://openlake/spark/sample-data/taxi-data.csv\"))\n",
    "```\n",
    "\n",
    "We transform the dataframe to avor in this code block\n",
    "\n",
    "```python\n",
    "df = df.select(to_avro(struct([df[x] for x in df.columns]), value_schema_str).alias(\"value\"))\n",
    "```\n",
    "\n",
    "Finally we write the Kafka events for the topic `nyc-avro-topic` using the following block to `my-kafka-cluster-kafka-bootstrap.kafka.svc.cluster.local:9092` kafka server and this `http://kafka-schema-registry-cp-schema-registry.kafka.svc.cluster.local:8081` kafka schema registry URL\n",
    "\n",
    "\n",
    "```python\n",
    "df.write \\\n",
    "    .format(\"kafka\") \\\n",
    "    .option(\"kafka.bootstrap.servers\", \n",
    "            os.getenv(\"KAFKA_BOOTSTRAM_SERVER\", \"my-kafka-cluster-kafka-bootstrap.kafka.svc.cluster.local:9092\")) \\\n",
    "    .option(\"flushInterval\", \"100ms\") \\\n",
    "    .option(\"key.serializer\", \"org.apache.kafka.common.serialization.StringSerializer\") \\\n",
    "    .option(\"value.serializer\", \"io.confluent.kafka.serializers.KafkaAvroSerializer\") \\\n",
    "    .option(\"schema.registry.url\",\n",
    "            os.getenv(\"KAFKA_SCHEMA_REGISTRY\", \"http://kafka-schema-registry-cp-schema-registry.kafka.svc.cluster.local:8081\")) \\\n",
    "    .option(\"topic\", \"nyc-avro-topic\") \\\n",
    "    .save()\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4f8a0bec",
   "metadata": {},
   "source": [
    "Build the Docker image with the above code or use `openlake/sparkjob-demo:3.3.2`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "e89b7a0c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting sample-code/spark-job/sparkjob-kafka-producer.yaml\n"
     ]
    }
   ],
   "source": [
    "%%writefile sample-code/spark-job/sparkjob-kafka-producer.yaml\n",
    "apiVersion: \"sparkoperator.k8s.io/v1beta2\"\n",
    "kind: SparkApplication\n",
    "metadata:\n",
    "  name: kafka-stream-producer\n",
    "  namespace: spark-operator\n",
    "spec:\n",
    "  type: Python\n",
    "  pythonVersion: \"3\"\n",
    "  mode: cluster\n",
    "  image: \"openlake/sparkjob-demo:3.3.2\"\n",
    "  imagePullPolicy: Always\n",
    "  mainApplicationFile: local:///app/spark-streaming-kafka-producer.py\n",
    "  sparkVersion: \"3.3.2\"\n",
    "  restartPolicy:\n",
    "    type: OnFailure\n",
    "    onFailureRetries: 3\n",
    "    onFailureRetryInterval: 10\n",
    "    onSubmissionFailureRetries: 5\n",
    "    onSubmissionFailureRetryInterval: 20\n",
    "  driver:\n",
    "    cores: 3\n",
    "    # coreLimit: \"1200m\"\n",
    "    memory: \"2048m\"\n",
    "    labels:\n",
    "      version: 3.3.2\n",
    "    serviceAccount: my-release-spark\n",
    "    env:\n",
    "      - name: INPUT_PATH\n",
    "        value: \"s3a://openlake/spark/sample-data/taxi-data.csv\"\n",
    "      - name: AWS_REGION\n",
    "        value: us-east-1\n",
    "      - name: AWS_ACCESS_KEY_ID\n",
    "        value: openlakeuser\n",
    "      - name: AWS_SECRET_ACCESS_KEY\n",
    "        value: openlakeuser\n",
    "  executor:\n",
    "    cores: 1\n",
    "    instances: 10\n",
    "    memory: \"1024m\"\n",
    "    labels:\n",
    "      version: 3.3.2\n",
    "    env:\n",
    "      - name: INPUT_PATH\n",
    "        value: \"s3a://openlake/spark/sample-data/taxi-data.csv\"\n",
    "      - name: AWS_REGION\n",
    "        value: us-east-1\n",
    "      - name: AWS_ACCESS_KEY_ID\n",
    "        value: openlakeuser\n",
    "      - name: AWS_SECRET_ACCESS_KEY\n",
    "        value: openlakeuser\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "56c6afe8",
   "metadata": {},
   "outputs": [],
   "source": [
    "!kubectl apply -f sample-code/spark-job/sparkjob-kafka-producer.yaml"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8024c516",
   "metadata": {},
   "source": [
    "Based on the configs we have setup with 10 partitions for `nyc-avro-topic` and 10 executors from spark Kafka prodcucer and consumer all the `~112M` rows is streamed and consumed in less than 10mins from the original ~3hrs from the previous kafka/spark streaming notebooks. This is a big performance gain and can be crucial based on the type of application that you want to build. \n",
    "\n",
    "**Note**: If we did not apply the `10 partitions` change in the `nyc-avro-topic` spark producer will still be able to complete in `<10 min` but the consumer will still take time to complete based on the number of partitions in the topic level."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "71ffc73f",
   "metadata": {},
   "source": [
    "### Performance Benchmarks\n",
    "\n",
    "If we measured the number of S3 API calls made by Spark structured streaming consumer alone we will get the below numbers\n",
    "\n",
    "\n",
    "```\n",
    "API                             RX      TX      CALLS   ERRORS\n",
    "s3.CompleteMultipartUpload      7.5 KiB 7.3 KiB 16      0       \n",
    "s3.DeleteMultipleObjects        22 KiB  6.8 KiB 60      0       \n",
    "s3.HeadObject                   51 KiB  0 B     200     0       \n",
    "s3.ListObjectsV1                1.5 KiB 9.0 KiB 6       0       \n",
    "s3.ListObjectsV2                15 KiB  20 KiB  60      0       \n",
    "s3.NewMultipartUpload           4.1 KiB 6.1 KiB 16      0       \n",
    "s3.PutObject                    1.1 GiB 0 B     63      0       \n",
    "s3.PutObjectPart                1.2 GiB 0 B     32      0       \n",
    "\n",
    "Summary:\n",
    "\n",
    "Total: 453 CALLS, 2.4 GiB RX, 49 KiB TX - in 160.71s\n",
    "```\n",
    "\n",
    "The same if done without the MinIO's checkpoint mananger we will end up with the below numbers\n",
    "\n",
    "```\n",
    "API                             RX      TX      CALLS   ERRORS\n",
    "s3.CompleteMultipartUpload      6.1 KiB 5.9 KiB 13      0       \n",
    "s3.CopyObject                   6.1 KiB 4.1 KiB 18      0       \n",
    "s3.DeleteMultipleObjects        30 KiB  8.8 KiB 78      0       \n",
    "s3.DeleteObject                 4.6 KiB 0 B     18      0       \n",
    "s3.HeadObject                   110 KiB 0 B     432     0       \n",
    "s3.ListObjectsV2                63 KiB  124 KiB 248     0       \n",
    "s3.NewMultipartUpload           3.3 KiB 4.9 KiB 13      0       \n",
    "s3.PutObject                    1.3 GiB 0 B     66      0       \n",
    "s3.PutObjectPart                1.1 GiB 0 B     26      0       \n",
    "\n",
    "Summary:\n",
    "\n",
    "Total: 912 CALLS, 2.3 GiB RX, 147 KiB TX - in 166.55s\n",
    "```\n",
    "\n",
    "We can clearly see the signigicant performance improvements with the end-to-end Spark structured streaming for kafka producer and consumer and with `MinIO's checkpoint manager` we further enhanced the performance by reducing the number of S3 API calls. \n",
    "\n",
    "Further if the above sample code was run on a `versioned bucket` and if we did a `mc ls --versions --recursive opl/warehouse-v/k8 --summarize` we will end up with `84 objects` vs `140 objects` between MinIO's checkpoint manager vs the default checkpoint manager respectively which again doesn't cleanup the delete markers."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5369d98e",
   "metadata": {},
   "source": []
  }
 ],
 "metadata": {
  "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.9.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}