1.Kafka config and data simulation to topic.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "8cb5ea75-5f71-4acf-b182-cd2c72ce02f0",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "1. API Setup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "d0925062-27e9-46e6-bd98-c376bab6c0fb",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "%python\n",
    "import requests\n",
    "import time\n",
    "\n",
    "# Alpha Vantage Configuration\n",
    "ALPHA_VANTAGE_API_KEY =  dbutils.secrets.get(scope=\"key_vault_scope\", key=\"MYAPIKEY\")\n",
    "SYMBOL = \"GOOGL\"\n",
    "INTERVAL = \"1min\"  # Possible values: 1min, 5min, 15min, 30min, 60min\n",
    "ALPHA_VANTAGE_API_URL = f\"https://www.alphavantage.co/query?function=TIME_SERIES_INTRADAY&symbol={SYMBOL}&interval={INTERVAL}&apikey={ALPHA_VANTAGE_API_KEY}\"\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "d7fb399b-e074-42eb-a35d-4670d8942658",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    " # Kafka setup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "15a9cfa0-eb20-4262-9c3c-cd0f9acb4431",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Collecting confluent_kafka\n  Obtaining dependency information for confluent_kafka from https://files.pythonhosted.org/packages/e0/37/4ab00b1562ba8f9b7a4e91fde2ce8b6105e9901a05f6852ffaf2fec44417/confluent_kafka-2.5.3-cp311-cp311-manylinux_2_28_x86_64.whl.metadata\n  Downloading confluent_kafka-2.5.3-cp311-cp311-manylinux_2_28_x86_64.whl.metadata (2.3 kB)\nDownloading confluent_kafka-2.5.3-cp311-cp311-manylinux_2_28_x86_64.whl (3.9 MB)\n\u001B[?25l   \u001B[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001B[0m \u001B[32m0.0/3.9 MB\u001B[0m \u001B[31m?\u001B[0m eta \u001B[36m-:--:--\u001B[0m\n\u001B[2K   \u001B[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001B[0m \u001B[32m0.0/3.9 MB\u001B[0m \u001B[31m1.3 MB/s\u001B[0m eta \u001B[36m0:00:04\u001B[0m\n\u001B[2K   \u001B[91m━\u001B[0m\u001B[90m╺\u001B[0m\u001B[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001B[0m \u001B[32m0.1/3.9 MB\u001B[0m \u001B[31m1.7 MB/s\u001B[0m eta \u001B[36m0:00:03\u001B[0m\n\u001B[2K   \u001B[91m━━━\u001B[0m\u001B[90m╺\u001B[0m\u001B[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001B[0m \u001B[32m0.3/3.9 MB\u001B[0m \u001B[31m2.9 MB/s\u001B[0m eta \u001B[36m0:00:02\u001B[0m\n\u001B[2K   \u001B[91m━━━━━━\u001B[0m\u001B[91m╸\u001B[0m\u001B[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001B[0m \u001B[32m0.7/3.9 MB\u001B[0m \u001B[31m5.0 MB/s\u001B[0m eta \u001B[36m0:00:01\u001B[0m\n\u001B[2K   \u001B[91m━━━━━━━━━━━━━━\u001B[0m\u001B[91m╸\u001B[0m\u001B[90m━━━━━━━━━━━━━━━━━━━━━━━━━\u001B[0m \u001B[32m1.5/3.9 MB\u001B[0m \u001B[31m8.7 MB/s\u001B[0m eta \u001B[36m0:00:01\u001B[0m\n\u001B[2K   \u001B[91m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001B[0m\u001B[90m╺\u001B[0m\u001B[90m━━━━━━\u001B[0m \u001B[32m3.3/3.9 MB\u001B[0m \u001B[31m16.0 MB/s\u001B[0m eta \u001B[36m0:00:01\u001B[0m\n\u001B[2K   \u001B[91m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001B[0m\u001B[91m╸\u001B[0m \u001B[32m3.9/3.9 MB\u001B[0m \u001B[31m18.4 MB/s\u001B[0m eta \u001B[36m0:00:01\u001B[0m\n\u001B[2K   \u001B[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001B[0m \u001B[32m3.9/3.9 MB\u001B[0m \u001B[31m14.8 MB/s\u001B[0m eta \u001B[36m0:00:00\u001B[0m\n\u001B[?25hInstalling collected packages: confluent_kafka\nSuccessfully installed confluent_kafka-2.5.3\n\u001B[43mNote: you may need to restart the kernel using %restart_python or dbutils.library.restartPython() to use updated packages.\u001B[0m\n"
     ]
    }
   ],
   "source": [
    "%python\n",
    "%pip install confluent_kafka"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "87734c02-9235-4378-9830-40a038ada1ba",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "%restart_python"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "3cc316bc-555c-41f6-91ec-4cf5adaee2fc",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "%python\n",
    "kafka_bootstrap_servers = dbutils.secrets.get(scope=\"key_vault_scope\", key=\"MYKAFKASERVER\")\n",
    "kafka_topic = dbutils.secrets.get(scope=\"key_vault_scope\", key=\"MYKAFKATOPIC\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "7c9a47fc-9bdc-4dfd-8ee0-2948d66b6b7d",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "# Kafka configuration and pulling data from API to Kafka topic"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "1742e814-952c-4915-92bb-50692bf170fb",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "\n",
    "# Import necessary libraries\n",
    "from confluent_kafka import Producer\n",
    "import requests\n",
    "import json\n",
    "import time\n",
    "\n",
    "# Kafka Configuration\n",
    "conf = {\n",
    "    'bootstrap.servers': 'pkc-56d1g.eastus.azure.confluent.cloud:9092',\n",
    "    'client.id': 'alpha-vantage-producer',\n",
    "    'security.protocol': 'SASL_SSL',\n",
    "    'sasl.mechanism': 'PLAIN',\n",
    "    'sasl.username': 'OPJG2TBO5A6V3KGQ',\n",
    "    'sasl.password': 'NktHxsYoNZZjBNQyfyBMsfqxo3FIe1soT6VhBiL82np1pTQtmqJwYvCzQTRULLVk'\n",
    "}\n",
    "\n",
    "# Initialize Kafka producer\n",
    "producer = Producer(conf)\n",
    "\n",
    "def delivery_report(err, msg):\n",
    "    \"\"\"Callback called once the message is delivered or delivery failed.\"\"\"\n",
    "    if err:\n",
    "        print(f\"Message delivery failed: {err}\")\n",
    "    else:\n",
    "        print(f\"Message delivered to {msg.topic()} [{msg.partition()}]\")\n",
    "\n",
    "def fetch_and_send_stock_data():\n",
    "    # Fetch stock data from Alpha Vantage API\n",
    "    response = requests.get(ALPHA_VANTAGE_API_URL)\n",
    "    \n",
    "    if response.status_code == 200:\n",
    "        stock_data = response.json()\n",
    "        # Extract time series data\n",
    "        time_series = stock_data.get(\"Time Series (1min)\", {})\n",
    "        for timestamp, data in time_series.items():\n",
    "            message = {\n",
    "                \"symbol\": SYMBOL,\n",
    "                \"timestamp\": timestamp,\n",
    "                \"open\": data[\"1. open\"],\n",
    "                \"high\": data[\"2. high\"],\n",
    "                \"low\": data[\"3. low\"],\n",
    "                \"close\": data[\"4. close\"],\n",
    "                \"volume\": data[\"5. volume\"]\n",
    "            }\n",
    "            # Send each data point to Kafka\n",
    "            producer.produce(\n",
    "                kafka_topic, \n",
    "                key=SYMBOL, \n",
    "                value=json.dumps(message), \n",
    "                callback=delivery_report\n",
    "            )\n",
    "            producer.flush()\n",
    "    else:\n",
    "        print(f\"Failed to fetch stock data: {response.status_code}\")\n",
    "\n",
    "# Call this function periodically (respecting Alpha Vantage rate limits)\n",
    "while True:\n",
    "    fetch_and_send_stock_data()\n",
    "    time.sleep(60)  # Sleep to respect the API rate limit (1 minute)"
   ]
  }
 ],
 "metadata": {
  "application/vnd.databricks.v1+notebook": {
   "dashboards": [],
   "environmentMetadata": null,
   "language": "python",
   "notebookMetadata": {
    "pythonIndentUnit": 4
   },
   "notebookName": "1.Kafka config and data simulation to topic",
   "widgets": {}
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}