KafkaConsumer.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 2,
   "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": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "consumer = KafkaConsumer(\n",
    "    'demo_test',\n",
    "     bootstrap_servers=[''],\n",
    "    value_deserializer=lambda x: loads(x.decode('utf-8')))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "ename": "KeyboardInterrupt",
     "evalue": "",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mKeyboardInterrupt\u001b[0m                         Traceback (most recent call last)",
      "Cell \u001b[0;32mIn[4], line 1\u001b[0m\n\u001b[0;32m----> 1\u001b[0m \u001b[39mfor\u001b[39;00m c \u001b[39min\u001b[39;00m consumer:\n\u001b[1;32m      2\u001b[0m     \u001b[39mprint\u001b[39m(c\u001b[39m.\u001b[39mvalue)\n",
      "File \u001b[0;32m/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/kafka/consumer/group.py:1193\u001b[0m, in \u001b[0;36mKafkaConsumer.__next__\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m   1191\u001b[0m     \u001b[39mreturn\u001b[39;00m \u001b[39mself\u001b[39m\u001b[39m.\u001b[39mnext_v1()\n\u001b[1;32m   1192\u001b[0m \u001b[39melse\u001b[39;00m:\n\u001b[0;32m-> 1193\u001b[0m     \u001b[39mreturn\u001b[39;00m \u001b[39mself\u001b[39;49m\u001b[39m.\u001b[39;49mnext_v2()\n",
      "File \u001b[0;32m/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/kafka/consumer/group.py:1201\u001b[0m, in \u001b[0;36mKafkaConsumer.next_v2\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m   1199\u001b[0m     \u001b[39mself\u001b[39m\u001b[39m.\u001b[39m_iterator \u001b[39m=\u001b[39m \u001b[39mself\u001b[39m\u001b[39m.\u001b[39m_message_generator_v2()\n\u001b[1;32m   1200\u001b[0m \u001b[39mtry\u001b[39;00m:\n\u001b[0;32m-> 1201\u001b[0m     \u001b[39mreturn\u001b[39;00m \u001b[39mnext\u001b[39;49m(\u001b[39mself\u001b[39;49m\u001b[39m.\u001b[39;49m_iterator)\n\u001b[1;32m   1202\u001b[0m \u001b[39mexcept\u001b[39;00m \u001b[39mStopIteration\u001b[39;00m:\n\u001b[1;32m   1203\u001b[0m     \u001b[39mself\u001b[39m\u001b[39m.\u001b[39m_iterator \u001b[39m=\u001b[39m \u001b[39mNone\u001b[39;00m\n",
      "File \u001b[0;32m/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/kafka/consumer/group.py:1116\u001b[0m, in \u001b[0;36mKafkaConsumer._message_generator_v2\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m   1114\u001b[0m \u001b[39mdef\u001b[39;00m \u001b[39m_message_generator_v2\u001b[39m(\u001b[39mself\u001b[39m):\n\u001b[1;32m   1115\u001b[0m     timeout_ms \u001b[39m=\u001b[39m \u001b[39m1000\u001b[39m \u001b[39m*\u001b[39m (\u001b[39mself\u001b[39m\u001b[39m.\u001b[39m_consumer_timeout \u001b[39m-\u001b[39m time\u001b[39m.\u001b[39mtime())\n\u001b[0;32m-> 1116\u001b[0m     record_map \u001b[39m=\u001b[39m \u001b[39mself\u001b[39;49m\u001b[39m.\u001b[39;49mpoll(timeout_ms\u001b[39m=\u001b[39;49mtimeout_ms, update_offsets\u001b[39m=\u001b[39;49m\u001b[39mFalse\u001b[39;49;00m)\n\u001b[1;32m   1117\u001b[0m     \u001b[39mfor\u001b[39;00m tp, records \u001b[39min\u001b[39;00m six\u001b[39m.\u001b[39miteritems(record_map):\n\u001b[1;32m   1118\u001b[0m         \u001b[39m# Generators are stateful, and it is possible that the tp / records\u001b[39;00m\n\u001b[1;32m   1119\u001b[0m         \u001b[39m# here may become stale during iteration -- i.e., we seek to a\u001b[39;00m\n\u001b[1;32m   1120\u001b[0m         \u001b[39m# different offset, pause consumption, or lose assignment.\u001b[39;00m\n\u001b[1;32m   1121\u001b[0m         \u001b[39mfor\u001b[39;00m record \u001b[39min\u001b[39;00m records:\n\u001b[1;32m   1122\u001b[0m             \u001b[39m# is_fetchable(tp) should handle assignment changes and offset\u001b[39;00m\n\u001b[1;32m   1123\u001b[0m             \u001b[39m# resets; for all other changes (e.g., seeks) we'll rely on the\u001b[39;00m\n\u001b[1;32m   1124\u001b[0m             \u001b[39m# outer function destroying the existing iterator/generator\u001b[39;00m\n\u001b[1;32m   1125\u001b[0m             \u001b[39m# via self._iterator = None\u001b[39;00m\n",
      "File \u001b[0;32m/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/kafka/consumer/group.py:655\u001b[0m, in \u001b[0;36mKafkaConsumer.poll\u001b[0;34m(self, timeout_ms, max_records, update_offsets)\u001b[0m\n\u001b[1;32m    653\u001b[0m remaining \u001b[39m=\u001b[39m timeout_ms\n\u001b[1;32m    654\u001b[0m \u001b[39mwhile\u001b[39;00m \u001b[39mTrue\u001b[39;00m:\n\u001b[0;32m--> 655\u001b[0m     records \u001b[39m=\u001b[39m \u001b[39mself\u001b[39;49m\u001b[39m.\u001b[39;49m_poll_once(remaining, max_records, update_offsets\u001b[39m=\u001b[39;49mupdate_offsets)\n\u001b[1;32m    656\u001b[0m     \u001b[39mif\u001b[39;00m records:\n\u001b[1;32m    657\u001b[0m         \u001b[39mreturn\u001b[39;00m records\n",
      "File \u001b[0;32m/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/kafka/consumer/group.py:702\u001b[0m, in \u001b[0;36mKafkaConsumer._poll_once\u001b[0;34m(self, timeout_ms, max_records, update_offsets)\u001b[0m\n\u001b[1;32m    699\u001b[0m     \u001b[39mself\u001b[39m\u001b[39m.\u001b[39m_client\u001b[39m.\u001b[39mpoll(timeout_ms\u001b[39m=\u001b[39m\u001b[39m0\u001b[39m)\n\u001b[1;32m    701\u001b[0m timeout_ms \u001b[39m=\u001b[39m \u001b[39mmin\u001b[39m(timeout_ms, \u001b[39mself\u001b[39m\u001b[39m.\u001b[39m_coordinator\u001b[39m.\u001b[39mtime_to_next_poll() \u001b[39m*\u001b[39m \u001b[39m1000\u001b[39m)\n\u001b[0;32m--> 702\u001b[0m \u001b[39mself\u001b[39;49m\u001b[39m.\u001b[39;49m_client\u001b[39m.\u001b[39;49mpoll(timeout_ms\u001b[39m=\u001b[39;49mtimeout_ms)\n\u001b[1;32m    703\u001b[0m \u001b[39m# after the long poll, we should check whether the group needs to rebalance\u001b[39;00m\n\u001b[1;32m    704\u001b[0m \u001b[39m# prior to returning data so that the group can stabilize faster\u001b[39;00m\n\u001b[1;32m    705\u001b[0m \u001b[39mif\u001b[39;00m \u001b[39mself\u001b[39m\u001b[39m.\u001b[39m_coordinator\u001b[39m.\u001b[39mneed_rejoin():\n",
      "File \u001b[0;32m/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/kafka/client_async.py:602\u001b[0m, in \u001b[0;36mKafkaClient.poll\u001b[0;34m(self, timeout_ms, future)\u001b[0m\n\u001b[1;32m    599\u001b[0m             timeout \u001b[39m=\u001b[39m \u001b[39mmin\u001b[39m(timeout, \u001b[39mself\u001b[39m\u001b[39m.\u001b[39mconfig[\u001b[39m'\u001b[39m\u001b[39mretry_backoff_ms\u001b[39m\u001b[39m'\u001b[39m])\n\u001b[1;32m    600\u001b[0m         timeout \u001b[39m=\u001b[39m \u001b[39mmax\u001b[39m(\u001b[39m0\u001b[39m, timeout)  \u001b[39m# avoid negative timeouts\u001b[39;00m\n\u001b[0;32m--> 602\u001b[0m     \u001b[39mself\u001b[39;49m\u001b[39m.\u001b[39;49m_poll(timeout \u001b[39m/\u001b[39;49m \u001b[39m1000\u001b[39;49m)\n\u001b[1;32m    604\u001b[0m \u001b[39m# called without the lock to avoid deadlock potential\u001b[39;00m\n\u001b[1;32m    605\u001b[0m \u001b[39m# if handlers need to acquire locks\u001b[39;00m\n\u001b[1;32m    606\u001b[0m responses\u001b[39m.\u001b[39mextend(\u001b[39mself\u001b[39m\u001b[39m.\u001b[39m_fire_pending_completed_requests())\n",
      "File \u001b[0;32m/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/site-packages/kafka/client_async.py:634\u001b[0m, in \u001b[0;36mKafkaClient._poll\u001b[0;34m(self, timeout)\u001b[0m\n\u001b[1;32m    631\u001b[0m \u001b[39mself\u001b[39m\u001b[39m.\u001b[39m_register_send_sockets()\n\u001b[1;32m    633\u001b[0m start_select \u001b[39m=\u001b[39m time\u001b[39m.\u001b[39mtime()\n\u001b[0;32m--> 634\u001b[0m ready \u001b[39m=\u001b[39m \u001b[39mself\u001b[39;49m\u001b[39m.\u001b[39;49m_selector\u001b[39m.\u001b[39;49mselect(timeout)\n\u001b[1;32m    635\u001b[0m end_select \u001b[39m=\u001b[39m time\u001b[39m.\u001b[39mtime()\n\u001b[1;32m    636\u001b[0m \u001b[39mif\u001b[39;00m \u001b[39mself\u001b[39m\u001b[39m.\u001b[39m_sensors:\n",
      "File \u001b[0;32m/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/selectors.py:561\u001b[0m, in \u001b[0;36mKqueueSelector.select\u001b[0;34m(self, timeout)\u001b[0m\n\u001b[1;32m    559\u001b[0m ready \u001b[39m=\u001b[39m []\n\u001b[1;32m    560\u001b[0m \u001b[39mtry\u001b[39;00m:\n\u001b[0;32m--> 561\u001b[0m     kev_list \u001b[39m=\u001b[39m \u001b[39mself\u001b[39m\u001b[39m.\u001b[39m_selector\u001b[39m.\u001b[39mcontrol(\u001b[39mNone\u001b[39;00m, max_ev, timeout)\n\u001b[1;32m    562\u001b[0m \u001b[39mexcept\u001b[39;00m \u001b[39mInterruptedError\u001b[39;00m:\n\u001b[1;32m    563\u001b[0m     \u001b[39mreturn\u001b[39;00m ready\n",
      "\u001b[0;31mKeyboardInterrupt\u001b[0m: "
     ]
    }
   ],
   "source": [
    "for c in consumer:\n",
    "    print(c.value)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "s3 = S3FileSystem()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for count, i in enumerate(consumer):\n",
    "    with s3.open(\"\".format(count), 'w') as file:\n",
    "        json.dump(i.value, file)  "
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.11.0 64-bit",
   "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.0"
  },
  "orig_nbformat": 4,
  "vscode": {
   "interpreter": {
    "hash": "1a1af0ee75eeea9e2e1ee996c87e7a2b11a0bebd85af04bb136d915cefc0abce"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}