OpenWeatherAPI_Kafka.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from confluent_kafka import Producer\n",
    "# from kafka import Producer\n",
    "\n",
    "import requests\n",
    "import json\n",
    "\n",
    "class WeatherKafkaProducer:\n",
    "    \"\"\"\n",
    "    Kafka Producer: Fetches weather data from OpenWeather API and publishes to Kafka.\n",
    "    \"\"\"\n",
    "    def __init__(self, kafka_broker, topic, api_key, city):\n",
    "        self.topic = topic\n",
    "        self.api_url = f\"http://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}\"\n",
    "        self.producer = Producer({'bootstrap.servers': kafka_broker})\n",
    "\n",
    "    def fetch_weather_data(self):\n",
    "        \"\"\"\n",
    "        Fetch weather data from OpenWeather API.\n",
    "        \"\"\"\n",
    "        try:\n",
    "            # Step 1: Send GET request to the OpenWeather API\n",
    "            response = requests.get(self.api_url)\n",
    "            # Step 2: Raise an exception for any non-2xx HTTP status code\n",
    "            response.raise_for_status()\n",
    "             # Step 3: Parse the response JSON and return it\n",
    "            return response.json()\n",
    "        except Exception as e:\n",
    "            # Step 4: Handle any errors that occurred during the API request\n",
    "            print(f\"Error fetching weather data: {e}\")\n",
    "            return None\n",
    "\n",
    "    def publish_to_kafka(self, data):\n",
    "        \"\"\"\n",
    "        Publish weather data to Kafka topic.\n",
    "        \"\"\"\n",
    "        try:\n",
    "            # Step 1: Produce message to Kafka topic\n",
    "            self.producer.produce(self.topic, key=\"weather\", value=json.dumps(data))\n",
    "            # Step 2: Flush the producer buffer (ensure all messages are sent)\n",
    "            self.producer.flush()\n",
    "            # Step 3: Print success message\n",
    "            print(f\"Published weather data to Kafka topic: {self.topic}\")\n",
    "        except Exception as e:\n",
    "            # Step 4: Handle any errors that occurred during message publishing\n",
    "            print(f\"Error publishing to Kafka: {e}\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from confluent_kafka import Consumer, KafkaException, KafkaError\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import col\n",
    "from pyspark.sql.types import StructType, StructField, StringType, DoubleType, IntegerType\n",
    "import json\n",
    "import os\n",
    "\n",
    "class KafkaSparkProcessor:\n",
    "    \"\"\"\n",
    "    Spark Processor: Consumes data from Kafka, processes it, and saves to CSV.\n",
    "    \"\"\"\n",
    "    def __init__(self, kafka_broker, group_id, topic, output_path):\n",
    "        self.topic = topic\n",
    "        self.output_path = output_path\n",
    "\n",
    "        # Initialize Spark Session\n",
    "        self.spark = SparkSession.builder \\\n",
    "            .appName(\"WeatherKafkaSparkProcessor\") \\\n",
    "            .getOrCreate()\n",
    "\n",
    "        # Initialize Kafka Consumer\n",
    "        self.consumer = Consumer({\n",
    "            'bootstrap.servers': kafka_broker,\n",
    "            'group.id': group_id,\n",
    "            'auto.offset.reset': 'earliest'\n",
    "        })\n",
    "        self.consumer.subscribe([self.topic])\n",
    "\n",
    "        # Define Weather Data Schema\n",
    "        self.weather_schema = StructType([\n",
    "            StructField(\"name\", StringType(), True),\n",
    "            StructField(\"main\", StructType([\n",
    "                StructField(\"temp\", DoubleType(), True),\n",
    "                StructField(\"pressure\", IntegerType(), True),\n",
    "                StructField(\"humidity\", IntegerType(), True)\n",
    "            ]), True),\n",
    "            StructField(\"wind\", StructType([\n",
    "                StructField(\"speed\", DoubleType(), True)\n",
    "            ]), True),\n",
    "            StructField(\"weather\", StructType([\n",
    "                StructField(\"description\", StringType(), True)\n",
    "            ]), True)\n",
    "        ])\n",
    "\n",
    "    def consume_and_process(self):\n",
    "        \"\"\"\n",
    "        Consume messages from Kafka, process with Spark, and save to CSV.\n",
    "        \"\"\"\n",
    "        messages = []\n",
    "        while True:\n",
    "            msg = self.consumer.poll(1.0)\n",
    "            if msg is None:\n",
    "                break\n",
    "            elif msg.error():\n",
    "                if msg.error().code() != KafkaError._PARTITION_EOF:\n",
    "                    raise KafkaException(msg.error())\n",
    "            else:\n",
    "                messages.append(msg.value().decode('utf-8'))\n",
    "\n",
    "        if messages:\n",
    "            # Convert JSON messages to Spark DataFrame\n",
    "            rdd = self.spark.sparkContext.parallelize(messages)\n",
    "            df = self.spark.read.json(rdd, schema=self.weather_schema)\n",
    "\n",
    "            # Extract and Transform Data\n",
    "            df_parsed = df.select(\n",
    "                col(\"name\").alias(\"city\"),\n",
    "                col(\"main.temp\").alias(\"temperature\"),\n",
    "                col(\"main.pressure\").alias(\"pressure\"),\n",
    "                col(\"main.humidity\").alias(\"humidity\"),\n",
    "                col(\"wind.speed\").alias(\"wind_speed\"),\n",
    "                col(\"weather.description\").alias(\"description\")\n",
    "            )\n",
    "\n",
    "            # Save to CSV\n",
    "            df_parsed.write.mode(\"append\").csv(self.output_path)\n",
    "            print(\"Processed and saved data to CSV.\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "\n",
    "class WeatherETLWorkflow:\n",
    "    \"\"\"\n",
    "    ETL Workflow: Coordinates Kafka Producer and Spark Processor.\n",
    "    \"\"\"\n",
    "    def __init__(self, kafka_broker, topic, api_key, city, output_path):\n",
    "        self.producer = WeatherKafkaProducer(kafka_broker, topic, api_key, city)\n",
    "        self.processor = KafkaSparkProcessor(kafka_broker, \"weather-consumer-group\", topic, output_path)\n",
    "\n",
    "    def run(self, interval=10):\n",
    "        \"\"\"\n",
    "        Run the ETL pipeline: Fetch data, produce to Kafka, process, and store to CSV.\n",
    "        \"\"\"\n",
    "        try:\n",
    "            while True:\n",
    "                # Step 1: Fetch and Publish Data\n",
    "                data = self.producer.fetch_weather_data()\n",
    "                if data:\n",
    "                    self.producer.publish_to_kafka(data)\n",
    "                \n",
    "                # Step 2: Consume and Process Data\n",
    "                self.processor.consume_and_process()\n",
    "\n",
    "                \n",
    "               # Wait for next iteration\n",
    "                time.sleep(interval)\n",
    "        except KeyboardInterrupt:\n",
    "            print(\"ETL workflow stopped by user.\")\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "24/12/22 22:14:40 WARN Utils: Your hostname, zhangmins-MacBook-Air.local resolves to a loopback address: 127.0.0.1; using 10.0.0.194 instead (on interface en0)\n",
      "24/12/22 22:14:40 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address\n",
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n",
      "24/12/22 22:14:42 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
      "24/12/22 22:14:43 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Published weather data to Kafka topic: weather_topic\n",
      "Published weather data to Kafka topic: weather_topic\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Processed and saved data to CSV.\n",
      "Published weather data to Kafka topic: weather_topic\n",
      "ETL workflow stopped by user.\n"
     ]
    }
   ],
   "source": [
    "\n",
    "if __name__ == \"__main__\":\n",
    "    # Configuration Parameters\n",
    "    KAFKA_BROKER = \"localhost:9092\"\n",
    "    TOPIC = \"weather_topic\"\n",
    "    API_KEY = \"89ecd29e95f497c6af8d6c94c73596d1\"\n",
    "    # CITY = \"London\"\n",
    "    CITY = \"Weihai\"\n",
    "    OUTPUT_PATH = \"data/real_time_weatherData_kafka.csv\"\n",
    "\n",
    "    # Initialize and Run Workflow\n",
    "    workflow = WeatherETLWorkflow(KAFKA_BROKER, TOPIC, API_KEY, CITY, OUTPUT_PATH)\n",
    "    workflow.run(interval=10)\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "base",
   "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.12.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}