KafkaConsumer.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "45bcb4bb-1428-4019-900b-be854d675bb2",
   "metadata": {},
   "outputs": [],
   "source": [
    "from kafka import KafkaConsumer\n",
    "from time import sleep\n",
    "from json import dumps,loads\n",
    "import json\n",
    "from s3fs import S3FileSystem"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e31d7518-187a-4552-a3f1-8e6e51ba45e3",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "0aa613f6-4482-4d7a-9157-b1f3624d01b4",
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "consumer = KafkaConsumer(\n",
    "    'demo_testing2',\n",
    "     bootstrap_servers=['44.201.147.56:9092'],\n",
    "    value_deserializer=lambda x: loads(x.decode('utf-8')))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "572ab04f-5e87-47e4-a94d-d9090a7f8fc2",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[{'Index': 'N225', 'Date': '1982-12-28', 'Open': 8016.669922, 'High': 8016.669922, 'Low': 8016.669922, 'Close': 8016.669922, 'Adj Close': 8016.669922, 'Volume': 0.0, 'CloseUSD': 80.16669922}]\n"
     ]
    },
    {
     "ename": "KeyboardInterrupt",
     "evalue": "",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mKeyboardInterrupt\u001b[0m                         Traceback (most recent call last)",
      "Cell \u001b[1;32mIn[4], line 1\u001b[0m\n\u001b[1;32m----> 1\u001b[0m \u001b[38;5;28;01mfor\u001b[39;00m c \u001b[38;5;129;01min\u001b[39;00m consumer:\n\u001b[0;32m      2\u001b[0m     \u001b[38;5;28mprint\u001b[39m(c\u001b[38;5;241m.\u001b[39mvalue)\n",
      "File \u001b[1;32mD:\\anaconda\\Lib\\site-packages\\kafka\\consumer\\group.py:1193\u001b[0m, in \u001b[0;36mKafkaConsumer.__next__\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m   1191\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mnext_v1()\n\u001b[0;32m   1192\u001b[0m \u001b[38;5;28;01melse\u001b[39;00m:\n\u001b[1;32m-> 1193\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mnext_v2()\n",
      "File \u001b[1;32mD:\\anaconda\\Lib\\site-packages\\kafka\\consumer\\group.py:1201\u001b[0m, in \u001b[0;36mKafkaConsumer.next_v2\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m   1199\u001b[0m     \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_iterator \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_message_generator_v2()\n\u001b[0;32m   1200\u001b[0m \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[1;32m-> 1201\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;28mnext\u001b[39m(\u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_iterator)\n\u001b[0;32m   1202\u001b[0m \u001b[38;5;28;01mexcept\u001b[39;00m \u001b[38;5;167;01mStopIteration\u001b[39;00m:\n\u001b[0;32m   1203\u001b[0m     \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_iterator \u001b[38;5;241m=\u001b[39m \u001b[38;5;28;01mNone\u001b[39;00m\n",
      "File \u001b[1;32mD:\\anaconda\\Lib\\site-packages\\kafka\\consumer\\group.py:1116\u001b[0m, in \u001b[0;36mKafkaConsumer._message_generator_v2\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m   1114\u001b[0m \u001b[38;5;28;01mdef\u001b[39;00m \u001b[38;5;21m_message_generator_v2\u001b[39m(\u001b[38;5;28mself\u001b[39m):\n\u001b[0;32m   1115\u001b[0m     timeout_ms \u001b[38;5;241m=\u001b[39m \u001b[38;5;241m1000\u001b[39m \u001b[38;5;241m*\u001b[39m (\u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_consumer_timeout \u001b[38;5;241m-\u001b[39m time\u001b[38;5;241m.\u001b[39mtime())\n\u001b[1;32m-> 1116\u001b[0m     record_map \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mpoll(timeout_ms\u001b[38;5;241m=\u001b[39mtimeout_ms, update_offsets\u001b[38;5;241m=\u001b[39m\u001b[38;5;28;01mFalse\u001b[39;00m)\n\u001b[0;32m   1117\u001b[0m     \u001b[38;5;28;01mfor\u001b[39;00m tp, records \u001b[38;5;129;01min\u001b[39;00m six\u001b[38;5;241m.\u001b[39miteritems(record_map):\n\u001b[0;32m   1118\u001b[0m         \u001b[38;5;66;03m# Generators are stateful, and it is possible that the tp / records\u001b[39;00m\n\u001b[0;32m   1119\u001b[0m         \u001b[38;5;66;03m# here may become stale during iteration -- i.e., we seek to a\u001b[39;00m\n\u001b[0;32m   1120\u001b[0m         \u001b[38;5;66;03m# different offset, pause consumption, or lose assignment.\u001b[39;00m\n\u001b[0;32m   1121\u001b[0m         \u001b[38;5;28;01mfor\u001b[39;00m record \u001b[38;5;129;01min\u001b[39;00m records:\n\u001b[0;32m   1122\u001b[0m             \u001b[38;5;66;03m# is_fetchable(tp) should handle assignment changes and offset\u001b[39;00m\n\u001b[0;32m   1123\u001b[0m             \u001b[38;5;66;03m# resets; for all other changes (e.g., seeks) we'll rely on the\u001b[39;00m\n\u001b[0;32m   1124\u001b[0m             \u001b[38;5;66;03m# outer function destroying the existing iterator/generator\u001b[39;00m\n\u001b[0;32m   1125\u001b[0m             \u001b[38;5;66;03m# via self._iterator = None\u001b[39;00m\n",
      "File \u001b[1;32mD:\\anaconda\\Lib\\site-packages\\kafka\\consumer\\group.py:655\u001b[0m, in \u001b[0;36mKafkaConsumer.poll\u001b[1;34m(self, timeout_ms, max_records, update_offsets)\u001b[0m\n\u001b[0;32m    653\u001b[0m remaining \u001b[38;5;241m=\u001b[39m timeout_ms\n\u001b[0;32m    654\u001b[0m \u001b[38;5;28;01mwhile\u001b[39;00m \u001b[38;5;28;01mTrue\u001b[39;00m:\n\u001b[1;32m--> 655\u001b[0m     records \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_poll_once(remaining, max_records, update_offsets\u001b[38;5;241m=\u001b[39mupdate_offsets)\n\u001b[0;32m    656\u001b[0m     \u001b[38;5;28;01mif\u001b[39;00m records:\n\u001b[0;32m    657\u001b[0m         \u001b[38;5;28;01mreturn\u001b[39;00m records\n",
      "File \u001b[1;32mD:\\anaconda\\Lib\\site-packages\\kafka\\consumer\\group.py:702\u001b[0m, in \u001b[0;36mKafkaConsumer._poll_once\u001b[1;34m(self, timeout_ms, max_records, update_offsets)\u001b[0m\n\u001b[0;32m    699\u001b[0m     \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_client\u001b[38;5;241m.\u001b[39mpoll(timeout_ms\u001b[38;5;241m=\u001b[39m\u001b[38;5;241m0\u001b[39m)\n\u001b[0;32m    701\u001b[0m timeout_ms \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mmin\u001b[39m(timeout_ms, \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_coordinator\u001b[38;5;241m.\u001b[39mtime_to_next_poll() \u001b[38;5;241m*\u001b[39m \u001b[38;5;241m1000\u001b[39m)\n\u001b[1;32m--> 702\u001b[0m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_client\u001b[38;5;241m.\u001b[39mpoll(timeout_ms\u001b[38;5;241m=\u001b[39mtimeout_ms)\n\u001b[0;32m    703\u001b[0m \u001b[38;5;66;03m# after the long poll, we should check whether the group needs to rebalance\u001b[39;00m\n\u001b[0;32m    704\u001b[0m \u001b[38;5;66;03m# prior to returning data so that the group can stabilize faster\u001b[39;00m\n\u001b[0;32m    705\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_coordinator\u001b[38;5;241m.\u001b[39mneed_rejoin():\n",
      "File \u001b[1;32mD:\\anaconda\\Lib\\site-packages\\kafka\\client_async.py:602\u001b[0m, in \u001b[0;36mKafkaClient.poll\u001b[1;34m(self, timeout_ms, future)\u001b[0m\n\u001b[0;32m    599\u001b[0m             timeout \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mmin\u001b[39m(timeout, \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mconfig[\u001b[38;5;124m'\u001b[39m\u001b[38;5;124mretry_backoff_ms\u001b[39m\u001b[38;5;124m'\u001b[39m])\n\u001b[0;32m    600\u001b[0m         timeout \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mmax\u001b[39m(\u001b[38;5;241m0\u001b[39m, timeout)  \u001b[38;5;66;03m# avoid negative timeouts\u001b[39;00m\n\u001b[1;32m--> 602\u001b[0m     \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_poll(timeout \u001b[38;5;241m/\u001b[39m \u001b[38;5;241m1000\u001b[39m)\n\u001b[0;32m    604\u001b[0m \u001b[38;5;66;03m# called without the lock to avoid deadlock potential\u001b[39;00m\n\u001b[0;32m    605\u001b[0m \u001b[38;5;66;03m# if handlers need to acquire locks\u001b[39;00m\n\u001b[0;32m    606\u001b[0m responses\u001b[38;5;241m.\u001b[39mextend(\u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_fire_pending_completed_requests())\n",
      "File \u001b[1;32mD:\\anaconda\\Lib\\site-packages\\kafka\\client_async.py:634\u001b[0m, in \u001b[0;36mKafkaClient._poll\u001b[1;34m(self, timeout)\u001b[0m\n\u001b[0;32m    631\u001b[0m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_register_send_sockets()\n\u001b[0;32m    633\u001b[0m start_select \u001b[38;5;241m=\u001b[39m time\u001b[38;5;241m.\u001b[39mtime()\n\u001b[1;32m--> 634\u001b[0m ready \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_selector\u001b[38;5;241m.\u001b[39mselect(timeout)\n\u001b[0;32m    635\u001b[0m end_select \u001b[38;5;241m=\u001b[39m time\u001b[38;5;241m.\u001b[39mtime()\n\u001b[0;32m    636\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_sensors:\n",
      "File \u001b[1;32mD:\\anaconda\\Lib\\selectors.py:323\u001b[0m, in \u001b[0;36mSelectSelector.select\u001b[1;34m(self, timeout)\u001b[0m\n\u001b[0;32m    321\u001b[0m ready \u001b[38;5;241m=\u001b[39m []\n\u001b[0;32m    322\u001b[0m \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[1;32m--> 323\u001b[0m     r, w, _ \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_select(\u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_readers, \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_writers, [], timeout)\n\u001b[0;32m    324\u001b[0m \u001b[38;5;28;01mexcept\u001b[39;00m \u001b[38;5;167;01mInterruptedError\u001b[39;00m:\n\u001b[0;32m    325\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m ready\n",
      "File \u001b[1;32mD:\\anaconda\\Lib\\selectors.py:314\u001b[0m, in \u001b[0;36mSelectSelector._select\u001b[1;34m(self, r, w, _, timeout)\u001b[0m\n\u001b[0;32m    313\u001b[0m \u001b[38;5;28;01mdef\u001b[39;00m \u001b[38;5;21m_select\u001b[39m(\u001b[38;5;28mself\u001b[39m, r, w, _, timeout\u001b[38;5;241m=\u001b[39m\u001b[38;5;28;01mNone\u001b[39;00m):\n\u001b[1;32m--> 314\u001b[0m     r, w, x \u001b[38;5;241m=\u001b[39m select\u001b[38;5;241m.\u001b[39mselect(r, w, w, timeout)\n\u001b[0;32m    315\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m r, w \u001b[38;5;241m+\u001b[39m x, []\n",
      "\u001b[1;31mKeyboardInterrupt\u001b[0m: "
     ]
    }
   ],
   "source": [
    "for c in consumer:\n",
    "    print(c.value)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "a6e74fb7-3560-4480-b9ef-e61b3e6d7d7d",
   "metadata": {},
   "outputs": [],
   "source": [
    "s3 = S3FileSystem()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "fd8f336e-82a8-49c2-8be4-dca89d425cdc",
   "metadata": {},
   "outputs": [
    {
     "ename": "JSONDecodeError",
     "evalue": "Expecting value: line 1 column 1 (char 0)",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mJSONDecodeError\u001b[0m                           Traceback (most recent call last)",
      "Cell \u001b[1;32mIn[6], line 1\u001b[0m\n\u001b[1;32m----> 1\u001b[0m \u001b[38;5;28;01mfor\u001b[39;00m count, i \u001b[38;5;129;01min\u001b[39;00m \u001b[38;5;28menumerate\u001b[39m(consumer):\n\u001b[0;32m      2\u001b[0m     \u001b[38;5;28;01mwith\u001b[39;00m s3\u001b[38;5;241m.\u001b[39mopen(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124ms3://kafka-stock-sam/stock_market_\u001b[39m\u001b[38;5;132;01m{}\u001b[39;00m\u001b[38;5;124m.json\u001b[39m\u001b[38;5;124m\"\u001b[39m\u001b[38;5;241m.\u001b[39mformat(count), \u001b[38;5;124m'\u001b[39m\u001b[38;5;124mw\u001b[39m\u001b[38;5;124m'\u001b[39m) \u001b[38;5;28;01mas\u001b[39;00m file:\n\u001b[0;32m      3\u001b[0m         json\u001b[38;5;241m.\u001b[39mdump(i\u001b[38;5;241m.\u001b[39mvalue, file)\n",
      "File \u001b[1;32mD:\\anaconda\\Lib\\site-packages\\kafka\\consumer\\group.py:1193\u001b[0m, in \u001b[0;36mKafkaConsumer.__next__\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m   1191\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mnext_v1()\n\u001b[0;32m   1192\u001b[0m \u001b[38;5;28;01melse\u001b[39;00m:\n\u001b[1;32m-> 1193\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mnext_v2()\n",
      "File \u001b[1;32mD:\\anaconda\\Lib\\site-packages\\kafka\\consumer\\group.py:1201\u001b[0m, in \u001b[0;36mKafkaConsumer.next_v2\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m   1199\u001b[0m     \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_iterator \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_message_generator_v2()\n\u001b[0;32m   1200\u001b[0m \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[1;32m-> 1201\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;28mnext\u001b[39m(\u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_iterator)\n\u001b[0;32m   1202\u001b[0m \u001b[38;5;28;01mexcept\u001b[39;00m \u001b[38;5;167;01mStopIteration\u001b[39;00m:\n\u001b[0;32m   1203\u001b[0m     \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_iterator \u001b[38;5;241m=\u001b[39m \u001b[38;5;28;01mNone\u001b[39;00m\n",
      "File \u001b[1;32mD:\\anaconda\\Lib\\site-packages\\kafka\\consumer\\group.py:1116\u001b[0m, in \u001b[0;36mKafkaConsumer._message_generator_v2\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m   1114\u001b[0m \u001b[38;5;28;01mdef\u001b[39;00m \u001b[38;5;21m_message_generator_v2\u001b[39m(\u001b[38;5;28mself\u001b[39m):\n\u001b[0;32m   1115\u001b[0m     timeout_ms \u001b[38;5;241m=\u001b[39m \u001b[38;5;241m1000\u001b[39m \u001b[38;5;241m*\u001b[39m (\u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_consumer_timeout \u001b[38;5;241m-\u001b[39m time\u001b[38;5;241m.\u001b[39mtime())\n\u001b[1;32m-> 1116\u001b[0m     record_map \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mpoll(timeout_ms\u001b[38;5;241m=\u001b[39mtimeout_ms, update_offsets\u001b[38;5;241m=\u001b[39m\u001b[38;5;28;01mFalse\u001b[39;00m)\n\u001b[0;32m   1117\u001b[0m     \u001b[38;5;28;01mfor\u001b[39;00m tp, records \u001b[38;5;129;01min\u001b[39;00m six\u001b[38;5;241m.\u001b[39miteritems(record_map):\n\u001b[0;32m   1118\u001b[0m         \u001b[38;5;66;03m# Generators are stateful, and it is possible that the tp / records\u001b[39;00m\n\u001b[0;32m   1119\u001b[0m         \u001b[38;5;66;03m# here may become stale during iteration -- i.e., we seek to a\u001b[39;00m\n\u001b[0;32m   1120\u001b[0m         \u001b[38;5;66;03m# different offset, pause consumption, or lose assignment.\u001b[39;00m\n\u001b[0;32m   1121\u001b[0m         \u001b[38;5;28;01mfor\u001b[39;00m record \u001b[38;5;129;01min\u001b[39;00m records:\n\u001b[0;32m   1122\u001b[0m             \u001b[38;5;66;03m# is_fetchable(tp) should handle assignment changes and offset\u001b[39;00m\n\u001b[0;32m   1123\u001b[0m             \u001b[38;5;66;03m# resets; for all other changes (e.g., seeks) we'll rely on the\u001b[39;00m\n\u001b[0;32m   1124\u001b[0m             \u001b[38;5;66;03m# outer function destroying the existing iterator/generator\u001b[39;00m\n\u001b[0;32m   1125\u001b[0m             \u001b[38;5;66;03m# via self._iterator = None\u001b[39;00m\n",
      "File \u001b[1;32mD:\\anaconda\\Lib\\site-packages\\kafka\\consumer\\group.py:655\u001b[0m, in \u001b[0;36mKafkaConsumer.poll\u001b[1;34m(self, timeout_ms, max_records, update_offsets)\u001b[0m\n\u001b[0;32m    653\u001b[0m remaining \u001b[38;5;241m=\u001b[39m timeout_ms\n\u001b[0;32m    654\u001b[0m \u001b[38;5;28;01mwhile\u001b[39;00m \u001b[38;5;28;01mTrue\u001b[39;00m:\n\u001b[1;32m--> 655\u001b[0m     records \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_poll_once(remaining, max_records, update_offsets\u001b[38;5;241m=\u001b[39mupdate_offsets)\n\u001b[0;32m    656\u001b[0m     \u001b[38;5;28;01mif\u001b[39;00m records:\n\u001b[0;32m    657\u001b[0m         \u001b[38;5;28;01mreturn\u001b[39;00m records\n",
      "File \u001b[1;32mD:\\anaconda\\Lib\\site-packages\\kafka\\consumer\\group.py:708\u001b[0m, in \u001b[0;36mKafkaConsumer._poll_once\u001b[1;34m(self, timeout_ms, max_records, update_offsets)\u001b[0m\n\u001b[0;32m    705\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_coordinator\u001b[38;5;241m.\u001b[39mneed_rejoin():\n\u001b[0;32m    706\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m {}\n\u001b[1;32m--> 708\u001b[0m records, _ \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_fetcher\u001b[38;5;241m.\u001b[39mfetched_records(max_records, update_offsets\u001b[38;5;241m=\u001b[39mupdate_offsets)\n\u001b[0;32m    709\u001b[0m \u001b[38;5;28;01mreturn\u001b[39;00m records\n",
      "File \u001b[1;32mD:\\anaconda\\Lib\\site-packages\\kafka\\consumer\\fetcher.py:344\u001b[0m, in \u001b[0;36mFetcher.fetched_records\u001b[1;34m(self, max_records, update_offsets)\u001b[0m\n\u001b[0;32m    342\u001b[0m         \u001b[38;5;28;01mbreak\u001b[39;00m\n\u001b[0;32m    343\u001b[0m     completion \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_completed_fetches\u001b[38;5;241m.\u001b[39mpopleft()\n\u001b[1;32m--> 344\u001b[0m     \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_next_partition_records \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_parse_fetched_data(completion)\n\u001b[0;32m    345\u001b[0m \u001b[38;5;28;01melse\u001b[39;00m:\n\u001b[0;32m    346\u001b[0m     records_remaining \u001b[38;5;241m-\u001b[39m\u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_append(drained,\n\u001b[0;32m    347\u001b[0m                                       \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_next_partition_records,\n\u001b[0;32m    348\u001b[0m                                       records_remaining,\n\u001b[0;32m    349\u001b[0m                                       update_offsets)\n",
      "File \u001b[1;32mD:\\anaconda\\Lib\\site-packages\\kafka\\consumer\\fetcher.py:818\u001b[0m, in \u001b[0;36mFetcher._parse_fetched_data\u001b[1;34m(self, completed_fetch)\u001b[0m\n\u001b[0;32m    814\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m records\u001b[38;5;241m.\u001b[39mhas_next():\n\u001b[0;32m    815\u001b[0m     log\u001b[38;5;241m.\u001b[39mdebug(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mAdding fetched record for partition \u001b[39m\u001b[38;5;132;01m%s\u001b[39;00m\u001b[38;5;124m with\u001b[39m\u001b[38;5;124m\"\u001b[39m\n\u001b[0;32m    816\u001b[0m               \u001b[38;5;124m\"\u001b[39m\u001b[38;5;124m offset \u001b[39m\u001b[38;5;132;01m%d\u001b[39;00m\u001b[38;5;124m to buffered record list\u001b[39m\u001b[38;5;124m\"\u001b[39m, tp,\n\u001b[0;32m    817\u001b[0m               position)\n\u001b[1;32m--> 818\u001b[0m     unpacked \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mlist\u001b[39m(\u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_unpack_message_set(tp, records))\n\u001b[0;32m    819\u001b[0m     parsed_records \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mPartitionRecords(fetch_offset, tp, unpacked)\n\u001b[0;32m    820\u001b[0m     last_offset \u001b[38;5;241m=\u001b[39m unpacked[\u001b[38;5;241m-\u001b[39m\u001b[38;5;241m1\u001b[39m]\u001b[38;5;241m.\u001b[39moffset\n",
      "File \u001b[1;32mD:\\anaconda\\Lib\\site-packages\\kafka\\consumer\\fetcher.py:473\u001b[0m, in \u001b[0;36mFetcher._unpack_message_set\u001b[1;34m(self, tp, records)\u001b[0m\n\u001b[0;32m    469\u001b[0m value_size \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mlen\u001b[39m(record\u001b[38;5;241m.\u001b[39mvalue) \u001b[38;5;28;01mif\u001b[39;00m record\u001b[38;5;241m.\u001b[39mvalue \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;129;01mnot\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m \u001b[38;5;28;01melse\u001b[39;00m \u001b[38;5;241m-\u001b[39m\u001b[38;5;241m1\u001b[39m\n\u001b[0;32m    470\u001b[0m key \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_deserialize(\n\u001b[0;32m    471\u001b[0m     \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mconfig[\u001b[38;5;124m'\u001b[39m\u001b[38;5;124mkey_deserializer\u001b[39m\u001b[38;5;124m'\u001b[39m],\n\u001b[0;32m    472\u001b[0m     tp\u001b[38;5;241m.\u001b[39mtopic, record\u001b[38;5;241m.\u001b[39mkey)\n\u001b[1;32m--> 473\u001b[0m value \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_deserialize(\n\u001b[0;32m    474\u001b[0m     \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mconfig[\u001b[38;5;124m'\u001b[39m\u001b[38;5;124mvalue_deserializer\u001b[39m\u001b[38;5;124m'\u001b[39m],\n\u001b[0;32m    475\u001b[0m     tp\u001b[38;5;241m.\u001b[39mtopic, record\u001b[38;5;241m.\u001b[39mvalue)\n\u001b[0;32m    476\u001b[0m headers \u001b[38;5;241m=\u001b[39m record\u001b[38;5;241m.\u001b[39mheaders\n\u001b[0;32m    477\u001b[0m header_size \u001b[38;5;241m=\u001b[39m \u001b[38;5;28msum\u001b[39m(\n\u001b[0;32m    478\u001b[0m     \u001b[38;5;28mlen\u001b[39m(h_key\u001b[38;5;241m.\u001b[39mencode(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mutf-8\u001b[39m\u001b[38;5;124m\"\u001b[39m)) \u001b[38;5;241m+\u001b[39m (\u001b[38;5;28mlen\u001b[39m(h_val) \u001b[38;5;28;01mif\u001b[39;00m h_val \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;129;01mnot\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m \u001b[38;5;28;01melse\u001b[39;00m \u001b[38;5;241m0\u001b[39m) \u001b[38;5;28;01mfor\u001b[39;00m h_key, h_val \u001b[38;5;129;01min\u001b[39;00m\n\u001b[0;32m    479\u001b[0m     headers) \u001b[38;5;28;01mif\u001b[39;00m headers \u001b[38;5;28;01melse\u001b[39;00m \u001b[38;5;241m-\u001b[39m\u001b[38;5;241m1\u001b[39m\n",
      "File \u001b[1;32mD:\\anaconda\\Lib\\site-packages\\kafka\\consumer\\fetcher.py:511\u001b[0m, in \u001b[0;36mFetcher._deserialize\u001b[1;34m(self, f, topic, bytes_)\u001b[0m\n\u001b[0;32m    509\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28misinstance\u001b[39m(f, Deserializer):\n\u001b[0;32m    510\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m f\u001b[38;5;241m.\u001b[39mdeserialize(topic, bytes_)\n\u001b[1;32m--> 511\u001b[0m \u001b[38;5;28;01mreturn\u001b[39;00m f(bytes_)\n",
      "Cell \u001b[1;32mIn[3], line 4\u001b[0m, in \u001b[0;36m<lambda>\u001b[1;34m(x)\u001b[0m\n\u001b[0;32m      1\u001b[0m consumer \u001b[38;5;241m=\u001b[39m KafkaConsumer(\n\u001b[0;32m      2\u001b[0m     \u001b[38;5;124m'\u001b[39m\u001b[38;5;124mdemo_testing2\u001b[39m\u001b[38;5;124m'\u001b[39m,\n\u001b[0;32m      3\u001b[0m      bootstrap_servers\u001b[38;5;241m=\u001b[39m[\u001b[38;5;124m'\u001b[39m\u001b[38;5;124m44.201.147.56:9092\u001b[39m\u001b[38;5;124m'\u001b[39m],\n\u001b[1;32m----> 4\u001b[0m     value_deserializer\u001b[38;5;241m=\u001b[39m\u001b[38;5;28;01mlambda\u001b[39;00m x: loads(x\u001b[38;5;241m.\u001b[39mdecode(\u001b[38;5;124m'\u001b[39m\u001b[38;5;124mutf-8\u001b[39m\u001b[38;5;124m'\u001b[39m)))\n",
      "File \u001b[1;32mD:\\anaconda\\Lib\\json\\__init__.py:346\u001b[0m, in \u001b[0;36mloads\u001b[1;34m(s, cls, object_hook, parse_float, parse_int, parse_constant, object_pairs_hook, **kw)\u001b[0m\n\u001b[0;32m    341\u001b[0m     s \u001b[38;5;241m=\u001b[39m s\u001b[38;5;241m.\u001b[39mdecode(detect_encoding(s), \u001b[38;5;124m'\u001b[39m\u001b[38;5;124msurrogatepass\u001b[39m\u001b[38;5;124m'\u001b[39m)\n\u001b[0;32m    343\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m (\u001b[38;5;28mcls\u001b[39m \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m \u001b[38;5;129;01mand\u001b[39;00m object_hook \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m \u001b[38;5;129;01mand\u001b[39;00m\n\u001b[0;32m    344\u001b[0m         parse_int \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m \u001b[38;5;129;01mand\u001b[39;00m parse_float \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m \u001b[38;5;129;01mand\u001b[39;00m\n\u001b[0;32m    345\u001b[0m         parse_constant \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m \u001b[38;5;129;01mand\u001b[39;00m object_pairs_hook \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m \u001b[38;5;129;01mand\u001b[39;00m \u001b[38;5;129;01mnot\u001b[39;00m kw):\n\u001b[1;32m--> 346\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m _default_decoder\u001b[38;5;241m.\u001b[39mdecode(s)\n\u001b[0;32m    347\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28mcls\u001b[39m \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m:\n\u001b[0;32m    348\u001b[0m     \u001b[38;5;28mcls\u001b[39m \u001b[38;5;241m=\u001b[39m JSONDecoder\n",
      "File \u001b[1;32mD:\\anaconda\\Lib\\json\\decoder.py:337\u001b[0m, in \u001b[0;36mJSONDecoder.decode\u001b[1;34m(self, s, _w)\u001b[0m\n\u001b[0;32m    332\u001b[0m \u001b[38;5;28;01mdef\u001b[39;00m \u001b[38;5;21mdecode\u001b[39m(\u001b[38;5;28mself\u001b[39m, s, _w\u001b[38;5;241m=\u001b[39mWHITESPACE\u001b[38;5;241m.\u001b[39mmatch):\n\u001b[0;32m    333\u001b[0m \u001b[38;5;250m    \u001b[39m\u001b[38;5;124;03m\"\"\"Return the Python representation of ``s`` (a ``str`` instance\u001b[39;00m\n\u001b[0;32m    334\u001b[0m \u001b[38;5;124;03m    containing a JSON document).\u001b[39;00m\n\u001b[0;32m    335\u001b[0m \n\u001b[0;32m    336\u001b[0m \u001b[38;5;124;03m    \"\"\"\u001b[39;00m\n\u001b[1;32m--> 337\u001b[0m     obj, end \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mraw_decode(s, idx\u001b[38;5;241m=\u001b[39m_w(s, \u001b[38;5;241m0\u001b[39m)\u001b[38;5;241m.\u001b[39mend())\n\u001b[0;32m    338\u001b[0m     end \u001b[38;5;241m=\u001b[39m _w(s, end)\u001b[38;5;241m.\u001b[39mend()\n\u001b[0;32m    339\u001b[0m     \u001b[38;5;28;01mif\u001b[39;00m end \u001b[38;5;241m!=\u001b[39m \u001b[38;5;28mlen\u001b[39m(s):\n",
      "File \u001b[1;32mD:\\anaconda\\Lib\\json\\decoder.py:355\u001b[0m, in \u001b[0;36mJSONDecoder.raw_decode\u001b[1;34m(self, s, idx)\u001b[0m\n\u001b[0;32m    353\u001b[0m     obj, end \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mscan_once(s, idx)\n\u001b[0;32m    354\u001b[0m \u001b[38;5;28;01mexcept\u001b[39;00m \u001b[38;5;167;01mStopIteration\u001b[39;00m \u001b[38;5;28;01mas\u001b[39;00m err:\n\u001b[1;32m--> 355\u001b[0m     \u001b[38;5;28;01mraise\u001b[39;00m JSONDecodeError(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mExpecting value\u001b[39m\u001b[38;5;124m\"\u001b[39m, s, err\u001b[38;5;241m.\u001b[39mvalue) \u001b[38;5;28;01mfrom\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m\n\u001b[0;32m    356\u001b[0m \u001b[38;5;28;01mreturn\u001b[39;00m obj, end\n",
      "\u001b[1;31mJSONDecodeError\u001b[0m: Expecting value: line 1 column 1 (char 0)"
     ]
    }
   ],
   "source": [
    "for count, i in enumerate(consumer):\n",
    "    with s3.open(\"s3://kafka-stock-sam/stock_market_{}.json\".format(count), 'w') as file:\n",
    "        json.dump(i.value, file)  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b6bf8861-1d32-4afa-9269-1731cf556c76",
   "metadata": {},
   "outputs": [],
   "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.11.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}