analysis.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/home/evan/Documents/hypersync-lancedb-pipe/.venv/lib/python3.12/site-packages/tqdm/auto.py:21: TqdmWarning: IProgress not found. Please update jupyter and ipywidgets. See https://ipywidgets.readthedocs.io/en/stable/user_install.html\n",
      "  from .autonotebook import tqdm as notebook_tqdm\n"
     ]
    }
   ],
   "source": [
    "import polars as pl\n",
    "import lancedb\n",
    "import time\n",
    "import duckdb"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Streaming Query Dataframe Output"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Run this notebook in a loop to monitor the progress of the dataset. Analytics can also be run simultaneously while the dataset is getting updated."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "19769573\n",
      "216257\n"
     ]
    },
    {
     "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[18], line 19\u001b[0m\n\u001b[1;32m     16\u001b[0m \u001b[38;5;66;03m# Get total number of rows\u001b[39;00m\n\u001b[1;32m     17\u001b[0m \u001b[38;5;28mprint\u001b[39m(table\u001b[38;5;241m.\u001b[39mto_polars()\u001b[38;5;241m.\u001b[39mselect(pl\u001b[38;5;241m.\u001b[39mlen())\u001b[38;5;241m.\u001b[39mcollect()\u001b[38;5;241m.\u001b[39mitem())\n\u001b[0;32m---> 19\u001b[0m \u001b[43mtime\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43msleep\u001b[49m\u001b[43m(\u001b[49m\u001b[38;5;241;43m180\u001b[39;49m\u001b[43m)\u001b[49m\n",
      "\u001b[0;31mKeyboardInterrupt\u001b[0m: "
     ]
    }
   ],
   "source": [
    "# continuously query the latest block number head while the dataset updates.\n",
    "while True:\n",
    "    table_name = \"blocks\"\n",
    "    db: lancedb.DBConnection = lancedb.connect(table_name)\n",
    "    # Try to open and merge data into existing table.\n",
    "    table: lancedb.table = db.open_table(table_name)\n",
    "\n",
    "    # Get the latest block number\n",
    "    print(\n",
    "        table.to_polars()\n",
    "        .select(\"block_number\")\n",
    "        .sort(by=\"block_number\", descending=True)\n",
    "        .collect()[\"block_number\"][0]\n",
    "    )\n",
    "\n",
    "    # Get total number of rows\n",
    "    print(table.to_polars().select(pl.len()).collect().item())\n",
    "\n",
    "    time.sleep(180)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### DuckDB Query Example"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "table_name = \"blocks\"\n",
    "db: lancedb.DBConnection = lancedb.connect(table_name)\n",
    "# open the table as a lance dataset to make it accessible for duckdb\n",
    "lance_dataset_table: lancedb.table = db.open_table(table_name).to_lance()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "┌───────────────────┐\n",
       "│ max(block_number) │\n",
       "│      uint64       │\n",
       "├───────────────────┤\n",
       "│          19769573 │\n",
       "└───────────────────┘"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# duckdb output format\n",
    "duckdb.sql(\"SELECT MAX(block_number) FROM lance_dataset_table\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div><style>\n",
       ".dataframe > thead > tr,\n",
       ".dataframe > tbody > tr {\n",
       "  text-align: right;\n",
       "  white-space: pre-wrap;\n",
       "}\n",
       "</style>\n",
       "<small>shape: (1, 1)</small><table border=\"1\" class=\"dataframe\"><thead><tr><th>max(block_number)</th></tr><tr><td>u64</td></tr></thead><tbody><tr><td>19769573</td></tr></tbody></table></div>"
      ],
      "text/plain": [
       "shape: (1, 1)\n",
       "┌───────────────────┐\n",
       "│ max(block_number) │\n",
       "│ ---               │\n",
       "│ u64               │\n",
       "╞═══════════════════╡\n",
       "│ 19769573          │\n",
       "└───────────────────┘"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# polars dataframe output format\n",
    "duckdb.sql(\"SELECT MAX(block_number) FROM lance_dataset_table\").pl()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".venv",
   "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
}