Team_13_Group_Project_Kafka_Consumer.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "cf89e964-2c0b-412c-b960-9a63d2e14b06",
   "metadata": {},
   "source": [
    "# Kafka Consumer Setup for Airbnb Data Flow"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4d6d9e1c-568b-4ae6-9931-403f50cf7a93",
   "metadata": {},
   "source": [
    "## Import Necessary Libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "02671465-46e6-4e53-8d16-82980a2b2990",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import col, expr, from_json\n",
    "from pyspark.sql.types import StructType, StructField, StringType, DoubleType, ArrayType, BooleanType, IntegerType, LongType, MapType"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9652a987-77a3-4fe5-9e37-06dfa9f55fb2",
   "metadata": {},
   "source": [
    "## Creating a Spark Session\n",
    "We'll create a spark session which connects with both local kafka and local mongodb"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "dee60bcd-7daa-445c-81e1-22fac1f49e6b",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "spark = SparkSession.builder \\\n",
    "    .appName(\"PySparkKafkaMongoDB\") \\\n",
    "    .master(\"local[1]\") \\\n",
    "    .config(\"spark.jars.packages\", \"org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.1,org.mongodb.spark:mongo-spark-connector_2.12:3.0.1\") \\\n",
    "    .config(\"spark.mongodb.output.uri\", \"mongodb://localhost:27017/project_budt737.airbnb_data\") \\\n",
    "    .getOrCreate()\n",
    "\n",
    "# Set log level to ERROR to reduce verbosity\n",
    "spark.sparkContext.setLogLevel(\"ERROR\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f80d39bb-aecf-4694-81a0-d485513a7fd0",
   "metadata": {},
   "source": [
    "## JSON Schema and Kafka Parameters"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9cfed176-f5c2-494f-b3a1-8831c6172c9b",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Define Kafka parameters\n",
    "kafka_bootstrap_servers = 'localhost:9092'\n",
    "kafka_topic_name = 'airbnb-test'\n",
    "\n",
    "schema = StructType([\n",
    "    StructField(\"id\", StringType(), True),\n",
    "    StructField(\"listing_url\", StringType(), True),\n",
    "    StructField(\"scrape_id\", StringType(), True),\n",
    "    StructField(\"last_scraped\", StringType(), True),\n",
    "    StructField(\"name\", StringType(), True),\n",
    "    StructField(\"summary\", StringType(), True),\n",
    "    StructField(\"space\", StringType(), True),\n",
    "    StructField(\"description\", StringType(), True),\n",
    "    StructField(\"experiences_offered\", StringType(), True),\n",
    "    StructField(\"neighborhood_overview\", StringType(), True),\n",
    "    StructField(\"notes\", StringType(), True),\n",
    "    StructField(\"transit\", StringType(), True),\n",
    "    StructField(\"access\", StringType(), True),\n",
    "    StructField(\"interaction\", StringType(), True),\n",
    "    StructField(\"house_rules\", StringType(), True),\n",
    "    StructField(\"thumbnail_url\", StringType(), True),\n",
    "    StructField(\"medium_url\", StringType(), True),\n",
    "    StructField(\"picture_url\", StringType(), True),\n",
    "    StructField(\"xl_picture_url\", StringType(), True),\n",
    "    StructField(\"host_id\", StringType(), True),\n",
    "    StructField(\"host_url\", StringType(), True),\n",
    "    StructField(\"host_name\", StringType(), True),\n",
    "    StructField(\"host_since\", StringType(), True),\n",
    "    StructField(\"host_location\", StringType(), True),\n",
    "    StructField(\"host_about\", StringType(), True),\n",
    "    StructField(\"host_response_time\", StringType(), True),\n",
    "    StructField(\"host_response_rate\", IntegerType(), True),\n",
    "    StructField(\"host_acceptance_rate\", StringType(), True),  # Assuming this can be null or a string\n",
    "    StructField(\"host_thumbnail_url\", StringType(), True),\n",
    "    StructField(\"host_picture_url\", StringType(), True),\n",
    "    StructField(\"host_neighbourhood\", StringType(), True),\n",
    "    StructField(\"host_listings_count\", IntegerType(), True),\n",
    "    StructField(\"host_total_listings_count\", IntegerType(), True),\n",
    "    StructField(\"host_verifications\", ArrayType(StringType()), True),\n",
    "    StructField(\"street\", StringType(), True),\n",
    "    StructField(\"neighbourhood\", StringType(), True),\n",
    "    StructField(\"neighbourhood_cleansed\", StringType(), True),\n",
    "    StructField(\"neighbourhood_group_cleansed\", StringType(), True),\n",
    "    StructField(\"city\", StringType(), True),\n",
    "    StructField(\"state\", StringType(), True),\n",
    "    StructField(\"zipcode\", StringType(), True),\n",
    "    StructField(\"market\", StringType(), True),\n",
    "    StructField(\"smart_location\", StringType(), True),\n",
    "    StructField(\"country_code\", StringType(), True),\n",
    "    StructField(\"country\", StringType(), True),\n",
    "    StructField(\"latitude\", DoubleType(), True),\n",
    "    StructField(\"longitude\", DoubleType(), True),\n",
    "    StructField(\"property_type\", StringType(), True),\n",
    "    StructField(\"room_type\", StringType(), True),\n",
    "    StructField(\"accommodates\", IntegerType(), True),\n",
    "    StructField(\"bathrooms\", DoubleType(), True),\n",
    "    StructField(\"bedrooms\", IntegerType(), True),\n",
    "    StructField(\"beds\", IntegerType(), True),\n",
    "    StructField(\"bed_type\", StringType(), True),\n",
    "    StructField(\"amenities\", ArrayType(StringType()), True),\n",
    "    StructField(\"square_feet\", StringType(), True),  # Assuming this can be null or a string\n",
    "    StructField(\"price\", IntegerType(), True),\n",
    "    StructField(\"weekly_price\", StringType(), True),  # Assuming this can be null or a string\n",
    "    StructField(\"monthly_price\", StringType(), True),  # Assuming this can be null or a string\n",
    "    StructField(\"security_deposit\", IntegerType(), True),\n",
    "    StructField(\"cleaning_fee\", IntegerType(), True),\n",
    "    StructField(\"guests_included\", IntegerType(), True),\n",
    "    StructField(\"extra_people\", IntegerType(), True),\n",
    "    StructField(\"minimum_nights\", IntegerType(), True),\n",
    "    StructField(\"maximum_nights\", IntegerType(), True),\n",
    "    StructField(\"calendar_updated\", StringType(), True),\n",
    "    StructField(\"has_availability\", StringType(), True),  # Assuming this can be null or a string\n",
    "    StructField(\"availability_30\", IntegerType(), True),\n",
    "    StructField(\"availability_60\", IntegerType(), True),\n",
    "    StructField(\"availability_90\", IntegerType(), True),\n",
    "    StructField(\"availability_365\", IntegerType(), True),\n",
    "    StructField(\"availability_365\", IntegerType(), True),\n",
    "    StructField(\"calendar_last_scraped\", StringType(), True),\n",
    "    StructField(\"number_of_reviews\", IntegerType(), True),\n",
    "    StructField(\"first_review\", StringType(), True),\n",
    "    StructField(\"last_review\", StringType(), True),\n",
    "    StructField(\"review_scores_rating\", IntegerType(), True),\n",
    "    StructField(\"review_scores_accuracy\", IntegerType(), True),\n",
    "    StructField(\"review_scores_cleanliness\", IntegerType(), True),\n",
    "    StructField(\"review_scores_checkin\", IntegerType(), True),\n",
    "    StructField(\"review_scores_communication\", IntegerType(), True),\n",
    "    StructField(\"review_scores_location\", IntegerType(), True),\n",
    "    StructField(\"review_scores_value\", IntegerType(), True),\n",
    "    StructField(\"license\", StringType(), True),\n",
    "    StructField(\"jurisdiction_names\", StringType(), True),  # Assuming single value; change if it's an array or a complex type\n",
    "    StructField(\"cancellation_policy\", StringType(), True),\n",
    "    StructField(\"calculated_host_listings_count\", IntegerType(), True),\n",
    "    StructField(\"reviews_per_month\", DoubleType(), True),\n",
    "    StructField(\"geolocation\", StructType([\n",
    "        StructField(\"lon\", DoubleType(), True),\n",
    "        StructField(\"lat\", DoubleType(), True)\n",
    "    ]), True),\n",
    "    StructField(\"features\", ArrayType(StringType()), True)\n",
    "])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e76e3c2a-5986-4f14-8137-fbcf500478da",
   "metadata": {},
   "source": [
    "## Reading from Kafka and writing to Mongo"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7a159515-1c1d-48a3-8831-98d02add6a02",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Read from Kafka\n",
    "df = spark.readStream.format(\"kafka\")\\\n",
    "    .option(\"kafka.bootstrap.servers\", kafka_bootstrap_servers)\\\n",
    "    .option(\"subscribe\", kafka_topic_name)\\\n",
    "    .option(\"startingOffsets\", \"latest\")\\\n",
    "    .load()\n",
    "\n",
    "df_string = df.selectExpr(\"CAST(value AS STRING)\")\n",
    "\n",
    "# Parse the JSON string\n",
    "df_structured = df_string.select(from_json(col(\"value\"), schema).alias(\"data\")).select(\"data.*\")\n",
    "\n",
    "df_transformed = df_structured.withColumnRenamed('id', '_id')\n",
    "\n",
    "# Write the structured stream to MongoDB\n",
    "def write_to_mongo(batch_df, epoch_id):\n",
    "    batch_df.write.format(\"mongo\").mode(\"append\").save()\n",
    "\n",
    "query = df_transformed.writeStream.foreachBatch(write_to_mongo).start()\n",
    "\n",
    "query.awaitTermination()"
   ]
  }
 ],
 "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.11.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}