{ "cells": [ { "cell_type": "code", "execution_count": 2, "id": "c77749d8", "metadata": {}, "outputs": [], "source": [ "from kafka import KafkaConsumer\n", "\n", "server = 'localhost:9092'\n", "topic_name = 'rides'" ] }, { "cell_type": "code", "execution_count": 3, "id": "74dcdffe", "metadata": {}, "outputs": [], "source": [ "from models import Ride, ride_deserializer" ] }, { "cell_type": "code", "execution_count": null, "id": "00726e41", "metadata": {}, "outputs": [], "source": [ "consumer = KafkaConsumer(\n", " topic_name,\n", " bootstrap_servers=[server],\n", " auto_offset_reset='earliest',\n", " group_id='rides-database',\n", " value_deserializer=ride_deserializer\n", ")" ] }, { "cell_type": "code", "execution_count": 1, "id": "a2cf7106", "metadata": {}, "outputs": [], "source": [ "import psycopg2\n", "\n", "conn = psycopg2.connect(\n", " host='localhost',\n", " port=5432,\n", " database='postgres',\n", " user='postgres',\n", " password='postgres'\n", ")\n", "\n", "conn.autocommit = True\n", "cur = conn.cursor()" ] }, { "cell_type": "code", "execution_count": 5, "id": "f0902406", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Listening to rides and writing to PostgreSQL...\n", "Inserted 100 rows...\n", "Inserted 200 rows...\n", "Inserted 300 rows...\n", "Inserted 400 rows...\n", "Inserted 500 rows...\n", "Inserted 600 rows...\n", "Inserted 700 rows...\n", "Inserted 800 rows...\n", "Inserted 900 rows...\n", "Inserted 1000 rows...\n" ] }, { "ename": "KeyboardInterrupt", "evalue": "", "output_type": "error", "traceback": [ "\u001b[31m---------------------------------------------------------------------------\u001b[39m", "\u001b[31mKeyboardInterrupt\u001b[39m Traceback (most recent call last)", "\u001b[36mCell\u001b[39m\u001b[36m \u001b[39m\u001b[32mIn[5]\u001b[39m\u001b[32m, line 6\u001b[39m\n\u001b[32m 3\u001b[39m \u001b[38;5;28mprint\u001b[39m(\u001b[33mf\u001b[39m\u001b[33m\"\u001b[39m\u001b[33mListening to \u001b[39m\u001b[38;5;132;01m{\u001b[39;00mtopic_name\u001b[38;5;132;01m}\u001b[39;00m\u001b[33m and writing to PostgreSQL...\u001b[39m\u001b[33m\"\u001b[39m)\n\u001b[32m 5\u001b[39m count = \u001b[32m0\u001b[39m\n\u001b[32m----> \u001b[39m\u001b[32m6\u001b[39m \u001b[38;5;28;43;01mfor\u001b[39;49;00m\u001b[43m \u001b[49m\u001b[43mmessage\u001b[49m\u001b[43m \u001b[49m\u001b[38;5;129;43;01min\u001b[39;49;00m\u001b[43m \u001b[49m\u001b[43mconsumer\u001b[49m\u001b[43m:\u001b[49m\n\u001b[32m 7\u001b[39m \u001b[43m \u001b[49m\u001b[43mride\u001b[49m\u001b[43m \u001b[49m\u001b[43m=\u001b[49m\u001b[43m \u001b[49m\u001b[43mmessage\u001b[49m\u001b[43m.\u001b[49m\u001b[43mvalue\u001b[49m\n\u001b[32m 8\u001b[39m \u001b[43m \u001b[49m\u001b[43mpickup_dt\u001b[49m\u001b[43m \u001b[49m\u001b[43m=\u001b[49m\u001b[43m \u001b[49m\u001b[43mdatetime\u001b[49m\u001b[43m.\u001b[49m\u001b[43mfromtimestamp\u001b[49m\u001b[43m(\u001b[49m\u001b[43mride\u001b[49m\u001b[43m.\u001b[49m\u001b[43mtpep_pickup_datetime\u001b[49m\u001b[43m \u001b[49m\u001b[43m/\u001b[49m\u001b[43m \u001b[49m\u001b[32;43m1000\u001b[39;49m\u001b[43m)\u001b[49m\n", "\u001b[36mFile \u001b[39m\u001b[32m/workspaces/streaming-workshop/.venv/lib/python3.12/site-packages/kafka/consumer/group.py:1213\u001b[39m, in \u001b[36mKafkaConsumer.__next__\u001b[39m\u001b[34m(self)\u001b[39m\n\u001b[32m 1211\u001b[39m \u001b[38;5;28mself\u001b[39m._iterator = \u001b[38;5;28mself\u001b[39m._message_generator_v2()\n\u001b[32m 1212\u001b[39m \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[32m-> \u001b[39m\u001b[32m1213\u001b[39m \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;28;43mnext\u001b[39;49m\u001b[43m(\u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43m_iterator\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 1214\u001b[39m \u001b[38;5;28;01mexcept\u001b[39;00m \u001b[38;5;167;01mStopIteration\u001b[39;00m:\n\u001b[32m 1215\u001b[39m \u001b[38;5;28mself\u001b[39m._iterator = \u001b[38;5;28;01mNone\u001b[39;00m\n", "\u001b[36mFile \u001b[39m\u001b[32m/workspaces/streaming-workshop/.venv/lib/python3.12/site-packages/kafka/consumer/group.py:1185\u001b[39m, in \u001b[36mKafkaConsumer._message_generator_v2\u001b[39m\u001b[34m(self)\u001b[39m\n\u001b[32m 1183\u001b[39m \u001b[38;5;28;01mdef\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[34m_message_generator_v2\u001b[39m(\u001b[38;5;28mself\u001b[39m):\n\u001b[32m 1184\u001b[39m timeout_ms = \u001b[32m1000\u001b[39m * \u001b[38;5;28mmax\u001b[39m(\u001b[32m0\u001b[39m, \u001b[38;5;28mself\u001b[39m._consumer_timeout - time.time())\n\u001b[32m-> \u001b[39m\u001b[32m1185\u001b[39m record_map = \u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43mpoll\u001b[49m\u001b[43m(\u001b[49m\u001b[43mtimeout_ms\u001b[49m\u001b[43m=\u001b[49m\u001b[43mtimeout_ms\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mupdate_offsets\u001b[49m\u001b[43m=\u001b[49m\u001b[38;5;28;43;01mFalse\u001b[39;49;00m\u001b[43m)\u001b[49m\n\u001b[32m 1186\u001b[39m \u001b[38;5;28;01mfor\u001b[39;00m tp, records \u001b[38;5;129;01min\u001b[39;00m six.iteritems(record_map):\n\u001b[32m 1187\u001b[39m \u001b[38;5;66;03m# Generators are stateful, and it is possible that the tp / records\u001b[39;00m\n\u001b[32m 1188\u001b[39m \u001b[38;5;66;03m# here may become stale during iteration -- i.e., we seek to a\u001b[39;00m\n\u001b[32m 1189\u001b[39m \u001b[38;5;66;03m# different offset, pause consumption, or lose assignment.\u001b[39;00m\n\u001b[32m 1190\u001b[39m \u001b[38;5;28;01mfor\u001b[39;00m record \u001b[38;5;129;01min\u001b[39;00m records:\n\u001b[32m 1191\u001b[39m \u001b[38;5;66;03m# is_fetchable(tp) should handle assignment changes and offset\u001b[39;00m\n\u001b[32m 1192\u001b[39m \u001b[38;5;66;03m# resets; for all other changes (e.g., seeks) we'll rely on the\u001b[39;00m\n\u001b[32m 1193\u001b[39m \u001b[38;5;66;03m# outer function destroying the existing iterator/generator\u001b[39;00m\n\u001b[32m 1194\u001b[39m \u001b[38;5;66;03m# via self._iterator = None\u001b[39;00m\n", "\u001b[36mFile \u001b[39m\u001b[32m/workspaces/streaming-workshop/.venv/lib/python3.12/site-packages/kafka/consumer/group.py:708\u001b[39m, in \u001b[36mKafkaConsumer.poll\u001b[39m\u001b[34m(self, timeout_ms, max_records, update_offsets)\u001b[39m\n\u001b[32m 706\u001b[39m timer = Timer(timeout_ms)\n\u001b[32m 707\u001b[39m \u001b[38;5;28;01mwhile\u001b[39;00m \u001b[38;5;129;01mnot\u001b[39;00m \u001b[38;5;28mself\u001b[39m._closed:\n\u001b[32m--> \u001b[39m\u001b[32m708\u001b[39m records = \u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43m_poll_once\u001b[49m\u001b[43m(\u001b[49m\u001b[43mtimer\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mmax_records\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mupdate_offsets\u001b[49m\u001b[43m=\u001b[49m\u001b[43mupdate_offsets\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 709\u001b[39m \u001b[38;5;28;01mif\u001b[39;00m records:\n\u001b[32m 710\u001b[39m \u001b[38;5;28;01mreturn\u001b[39;00m records\n", "\u001b[36mFile \u001b[39m\u001b[32m/workspaces/streaming-workshop/.venv/lib/python3.12/site-packages/kafka/consumer/group.py:757\u001b[39m, in \u001b[36mKafkaConsumer._poll_once\u001b[39m\u001b[34m(self, timer, max_records, update_offsets)\u001b[39m\n\u001b[32m 754\u001b[39m log.debug(\u001b[33m'\u001b[39m\u001b[33mpoll: do not have all fetch positions...\u001b[39m\u001b[33m'\u001b[39m)\n\u001b[32m 755\u001b[39m poll_timeout_ms = \u001b[38;5;28mmin\u001b[39m(poll_timeout_ms, \u001b[38;5;28mself\u001b[39m.config[\u001b[33m'\u001b[39m\u001b[33mretry_backoff_ms\u001b[39m\u001b[33m'\u001b[39m])\n\u001b[32m--> \u001b[39m\u001b[32m757\u001b[39m \u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43m_client\u001b[49m\u001b[43m.\u001b[49m\u001b[43mpoll\u001b[49m\u001b[43m(\u001b[49m\u001b[43mtimeout_ms\u001b[49m\u001b[43m=\u001b[49m\u001b[43mpoll_timeout_ms\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 758\u001b[39m \u001b[38;5;66;03m# after the long poll, we should check whether the group needs to rebalance\u001b[39;00m\n\u001b[32m 759\u001b[39m \u001b[38;5;66;03m# prior to returning data so that the group can stabilize faster\u001b[39;00m\n\u001b[32m 760\u001b[39m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28mself\u001b[39m._coordinator.need_rejoin():\n", "\u001b[36mFile \u001b[39m\u001b[32m/workspaces/streaming-workshop/.venv/lib/python3.12/site-packages/kafka/client_async.py:685\u001b[39m, in \u001b[36mKafkaClient.poll\u001b[39m\u001b[34m(self, timeout_ms, future)\u001b[39m\n\u001b[32m 678\u001b[39m timeout = \u001b[38;5;28mmin\u001b[39m(\n\u001b[32m 679\u001b[39m user_timeout_ms,\n\u001b[32m 680\u001b[39m metadata_timeout_ms,\n\u001b[32m 681\u001b[39m idle_connection_timeout_ms,\n\u001b[32m 682\u001b[39m request_timeout_ms)\n\u001b[32m 683\u001b[39m timeout = \u001b[38;5;28mmax\u001b[39m(\u001b[32m0\u001b[39m, timeout) \u001b[38;5;66;03m# avoid negative timeouts\u001b[39;00m\n\u001b[32m--> \u001b[39m\u001b[32m685\u001b[39m \u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43m_poll\u001b[49m\u001b[43m(\u001b[49m\u001b[43mtimeout\u001b[49m\u001b[43m \u001b[49m\u001b[43m/\u001b[49m\u001b[43m \u001b[49m\u001b[32;43m1000\u001b[39;49m\u001b[43m)\u001b[49m\n\u001b[32m 687\u001b[39m \u001b[38;5;66;03m# called without the lock to avoid deadlock potential\u001b[39;00m\n\u001b[32m 688\u001b[39m \u001b[38;5;66;03m# if handlers need to acquire locks\u001b[39;00m\n\u001b[32m 689\u001b[39m responses.extend(\u001b[38;5;28mself\u001b[39m._fire_pending_completed_requests())\n", "\u001b[36mFile \u001b[39m\u001b[32m/workspaces/streaming-workshop/.venv/lib/python3.12/site-packages/kafka/client_async.py:781\u001b[39m, in \u001b[36mKafkaClient._poll\u001b[39m\u001b[34m(self, timeout)\u001b[39m\n\u001b[32m 778\u001b[39m \u001b[38;5;28;01mcontinue\u001b[39;00m\n\u001b[32m 780\u001b[39m \u001b[38;5;28mself\u001b[39m._idle_expiry_manager.update(conn.node_id)\n\u001b[32m--> \u001b[39m\u001b[32m781\u001b[39m \u001b[38;5;28mself\u001b[39m._pending_completion.extend(\u001b[43mconn\u001b[49m\u001b[43m.\u001b[49m\u001b[43mrecv\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m)\n\u001b[32m 783\u001b[39m \u001b[38;5;66;03m# Check for additional pending SSL bytes\u001b[39;00m\n\u001b[32m 784\u001b[39m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28mself\u001b[39m.config[\u001b[33m'\u001b[39m\u001b[33msecurity_protocol\u001b[39m\u001b[33m'\u001b[39m] \u001b[38;5;129;01min\u001b[39;00m (\u001b[33m'\u001b[39m\u001b[33mSSL\u001b[39m\u001b[33m'\u001b[39m, \u001b[33m'\u001b[39m\u001b[33mSASL_SSL\u001b[39m\u001b[33m'\u001b[39m):\n\u001b[32m 785\u001b[39m \u001b[38;5;66;03m# TODO: optimize\u001b[39;00m\n", "\u001b[36mFile \u001b[39m\u001b[32m/workspaces/streaming-workshop/.venv/lib/python3.12/site-packages/kafka/conn.py:1131\u001b[39m, in \u001b[36mBrokerConnection.recv\u001b[39m\u001b[34m(self)\u001b[39m\n\u001b[32m 1126\u001b[39m \u001b[38;5;28;01mdef\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[34mrecv\u001b[39m(\u001b[38;5;28mself\u001b[39m):\n\u001b[32m 1127\u001b[39m \u001b[38;5;250m \u001b[39m\u001b[33;03m\"\"\"Non-blocking network receive.\u001b[39;00m\n\u001b[32m 1128\u001b[39m \n\u001b[32m 1129\u001b[39m \u001b[33;03m Return list of (response, future) tuples\u001b[39;00m\n\u001b[32m 1130\u001b[39m \u001b[33;03m \"\"\"\u001b[39;00m\n\u001b[32m-> \u001b[39m\u001b[32m1131\u001b[39m responses = \u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43m_recv\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 1132\u001b[39m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;129;01mnot\u001b[39;00m responses \u001b[38;5;129;01mand\u001b[39;00m \u001b[38;5;28mself\u001b[39m.requests_timed_out():\n\u001b[32m 1133\u001b[39m timed_out = \u001b[38;5;28mself\u001b[39m.timed_out_ifrs()\n", "\u001b[36mFile \u001b[39m\u001b[32m/workspaces/streaming-workshop/.venv/lib/python3.12/site-packages/kafka/conn.py:1202\u001b[39m, in \u001b[36mBrokerConnection._recv\u001b[39m\u001b[34m(self)\u001b[39m\n\u001b[32m 1200\u001b[39m recvd_data = \u001b[33mb\u001b[39m\u001b[33m'\u001b[39m\u001b[33m'\u001b[39m.join(recvd)\n\u001b[32m 1201\u001b[39m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28mself\u001b[39m._sensors:\n\u001b[32m-> \u001b[39m\u001b[32m1202\u001b[39m \u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43m_sensors\u001b[49m\u001b[43m.\u001b[49m\u001b[43mbytes_received\u001b[49m\u001b[43m.\u001b[49m\u001b[43mrecord\u001b[49m\u001b[43m(\u001b[49m\u001b[38;5;28;43mlen\u001b[39;49m\u001b[43m(\u001b[49m\u001b[43mrecvd_data\u001b[49m\u001b[43m)\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 1204\u001b[39m \u001b[38;5;66;03m# We need to keep the lock through protocol receipt\u001b[39;00m\n\u001b[32m 1205\u001b[39m \u001b[38;5;66;03m# so that we ensure that the processed byte order is the\u001b[39;00m\n\u001b[32m 1206\u001b[39m \u001b[38;5;66;03m# same as the received byte order\u001b[39;00m\n\u001b[32m 1207\u001b[39m \u001b[38;5;28;01mtry\u001b[39;00m:\n", "\u001b[36mFile \u001b[39m\u001b[32m/workspaces/streaming-workshop/.venv/lib/python3.12/site-packages/kafka/metrics/stats/sensor.py:77\u001b[39m, in \u001b[36mSensor.record\u001b[39m\u001b[34m(self, value, time_ms)\u001b[39m\n\u001b[32m 74\u001b[39m \u001b[38;5;28;01mwith\u001b[39;00m \u001b[38;5;28mself\u001b[39m._lock: \u001b[38;5;66;03m# XXX high volume, might be performance issue\u001b[39;00m\n\u001b[32m 75\u001b[39m \u001b[38;5;66;03m# increment all the stats\u001b[39;00m\n\u001b[32m 76\u001b[39m \u001b[38;5;28;01mfor\u001b[39;00m stat \u001b[38;5;129;01min\u001b[39;00m \u001b[38;5;28mself\u001b[39m._stats:\n\u001b[32m---> \u001b[39m\u001b[32m77\u001b[39m \u001b[43mstat\u001b[49m\u001b[43m.\u001b[49m\u001b[43mrecord\u001b[49m\u001b[43m(\u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[43m.\u001b[49m\u001b[43m_config\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mvalue\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mtime_ms\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m 78\u001b[39m \u001b[38;5;28mself\u001b[39m._check_quotas(time_ms)\n\u001b[32m 79\u001b[39m \u001b[38;5;28;01mfor\u001b[39;00m parent \u001b[38;5;129;01min\u001b[39;00m \u001b[38;5;28mself\u001b[39m._parents:\n", "\u001b[36mFile \u001b[39m\u001b[32m/workspaces/streaming-workshop/.venv/lib/python3.12/site-packages/kafka/metrics/stats/rate.py:49\u001b[39m, in \u001b[36mRate.record\u001b[39m\u001b[34m(self, config, value, time_ms)\u001b[39m\n\u001b[32m 46\u001b[39m \u001b[38;5;28;01mdef\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[34munit_name\u001b[39m(\u001b[38;5;28mself\u001b[39m):\n\u001b[32m 47\u001b[39m \u001b[38;5;28;01mreturn\u001b[39;00m TimeUnit.get_name(\u001b[38;5;28mself\u001b[39m._unit)\n\u001b[32m---> \u001b[39m\u001b[32m49\u001b[39m \u001b[38;5;28;01mdef\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[34mrecord\u001b[39m(\u001b[38;5;28mself\u001b[39m, config, value, time_ms):\n\u001b[32m 50\u001b[39m \u001b[38;5;28mself\u001b[39m._stat.record(config, value, time_ms)\n\u001b[32m 52\u001b[39m \u001b[38;5;28;01mdef\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[34mmeasure\u001b[39m(\u001b[38;5;28mself\u001b[39m, config, now):\n", "\u001b[31mKeyboardInterrupt\u001b[39m: " ] } ], "source": [ "from datetime import datetime\n", "\n", "print(f\"Listening to {topic_name} and writing to PostgreSQL...\")\n", "\n", "count = 0\n", "for message in consumer:\n", " ride = message.value\n", " pickup_dt = datetime.fromtimestamp(ride.tpep_pickup_datetime / 1000)\n", " cur.execute(\n", " \"\"\"INSERT INTO processed_events\n", " (PULocationID, DOLocationID, trip_distance, total_amount, pickup_datetime)\n", " VALUES (%s, %s, %s, %s, %s)\"\"\",\n", " (ride.PULocationID, ride.DOLocationID,\n", " ride.trip_distance, ride.total_amount, pickup_dt)\n", " )\n", " count += 1\n", " if count % 100 == 0:\n", " print(f\"Inserted {count} rows...\")\n", "\n", "consumer.close()\n", "cur.close()\n", "conn.close()" ] }, { "cell_type": "code", "execution_count": null, "id": "66840c80", "metadata": {}, "outputs": [], "source": [] }, { "cell_type": "code", "execution_count": null, "id": "2bec0472", "metadata": {}, "outputs": [], "source": [] } ], "metadata": { "kernelspec": { "display_name": "streaming-workshop", "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.1" } }, "nbformat": 4, "nbformat_minor": 5 }