ApacheSpark/02-streaming-kafka-src-dst-mongodb.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Below serve as example on how this script should look like and not represent the full script"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# import related libraries\n",
    "\n",
    "# Spark session & context\n",
    "spark = (SparkSession\n",
    "         .builder\n",
    "         .master('local')\n",
    "         .appName('kafka-mongo-streaming')     \n",
    "         .config(\"add spark, kafka,mongodb package here. ensure version is compatible\")\n",
    "         # Mongo config including the username and password from compose file\n",
    "         .config(\"enter mongodb uri here with related credentials\")\n",
    "         .getOrCreate())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Read the message from the kafka stream\n",
    "df = spark \\\n",
    "  .readStream \\\n",
    "  .format(\"kafka\") \\\n",
    "  .option(\"kafka.bootstrap.servers\", \"kafka related port\") \\\n",
    "  .option(\"subscribe\", \"created kafka topic\") \\\n",
    "  .load()\n",
    "\n",
    "# convert the binary values to string\n",
    "df1 = df.selectExpr(\"CAST(key AS STRING)\", \"CAST(value AS STRING)\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Create a temporary view for SparkSQL\n",
    "df1.createOrReplaceTempView(\"query\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<pyspark.sql.streaming.StreamingQuery at 0x7f1310608f90>"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Write out the message to the console of the environment\n",
    "res = spark.sql(\"your SQL statement\")\n",
    "res.writeStream.format(\"console\") \\\n",
    "            .outputMode(\"append\") \\\n",
    "            .start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import from_json, col, struct\n",
    "from pyspark.sql.types import StructType, StructField, StringType, DoubleType, IntegerType\n",
    "\n",
    "def foreach_batch_function(df, epoch_id):\n",
    "    # Define schemas for nested JSON fields\n",
    "    card_holder_schema = StructType([\n",
    "        StructField(\"first\", StringType(), True),\n",
    "        StructField(\"last\", StringType(), True),\n",
    "        StructField(\"gender\", StringType(), True),\n",
    "        StructField(\"dob\", StringType(), True)\n",
    "    ])\n",
    "    #Transform the values of all rows in column value and create a dataframe out of it (will also only have one row)\n",
    "    df2=df.withColumn(\"value\",from_json(df.value,\"do your transformation function here\"))\n",
    "    # Transform the dataframe so that it will include individual columns based on the BaseModel structure\n",
    "    df3 = df2.select(\n",
    "        col(\"value.trans_num\").alias(\"trans_num\"),\n",
    "        col(\"value.trans_date_trans_time\").alias(\"trans_date_trans_time\"),\n",
    "        col(\"value.cc_num\").alias(\"cc_num\"),\n",
    "        col(\"value.merchant\").alias(\"merchant\"),\n",
    "        col(\"value.category\").alias(\"category\"),\n",
    "        col(\"value.amt\").alias(\"amt\")\n",
    "    )\n",
    "    \n",
    "    # Transform the DataFrame\n",
    "    transformed_df = df3.select(\n",
    "        col(\"trans_num\"),\n",
    "        col(\"trans_date_trans_time\"),\n",
    "        col(\"cc_num\"),\n",
    "        col(\"merchant\"),\n",
    "        col(\"category\"),\n",
    "        col(\"amt\").cast(DoubleType())\n",
    "    )\n",
    "    \n",
    "    transformed_df.write.format(\"related mongodb spark format source\").mode(\"related mode to save json\").save()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "# use above function to run spark microbatch\n",
    "# Start the MongoDB stream and wait for termination"
   ]
  }
 ],
 "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.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}