Feature/delta inventory source

analysis/structured_metadata/live-table.ipynb

@@ -0,0 +1,113 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "5f890809",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import orjson\n",
+    "import genson\n",
+    "import natsort\n",
+    "import polars\n",
+    "import rich\n",
+    "from data_index.iceberg_config import S3TablesCatalogConfig, IcebergTableConfig\n",
+    "import pyiceberg\n",
+    "import datetime"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "f4126896",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# --- Sinks config ---\n",
+    "data_index_catalog_config = S3TablesCatalogConfig(\n",
+    "    region=\"ap-southeast-2\",\n",
+    "    arn=\"arn:aws:s3tables:ap-southeast-2:704910415367:bucket/imos-data-inventory\",\n",
+    ")\n",
+    "\n",
+    "table = IcebergTableConfig(\n",
+    "    catalog_config=data_index_catalog_config,\n",
+    "    namespace=\"inventory\",\n",
+    "    table_name=f\"live\",\n",
+    ").load()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "3cde5d68",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "table.schema"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "b8a717bf",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "def get_lookback_timestamp(time_delta: datetime.timedelta) -> str:\n",
+    "    return datetime.datetime.combine(\n",
+    "        date=datetime.date.today() - time_delta, \n",
+    "        time=datetime.time.min,\n",
+    "    ).isoformat()\n",
+    "\n",
+    "get_lookback_timestamp(time_delta=datetime.timedelta(days=10))"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "47537397",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "(\n",
+    "    table.scan(\n",
+    "        row_filter=f\"last_modified_date >= '{get_lookback_timestamp(time_delta=datetime.timedelta(days=10))}'\"\n",
+    "    )\n",
+    "    .to_polars()\n",
+    ")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "aff6129c",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "table.scan().to_polars()"
+   ]
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "data-index (3.12.13)",
+   "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.13"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}

analysis/structured_metadata/v5.ipynb

@@ -0,0 +1,183 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "77751b75",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import orjson\n",
+    "import genson\n",
+    "import natsort\n",
+    "import polars\n",
+    "import rich\n",
+    "from data_index.iceberg_config import S3TablesCatalogConfig, IcebergTableConfig"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "f40ca1d4",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# --- Sinks config ---\n",
+    "data_index_catalog_config = S3TablesCatalogConfig(\n",
+    "    region=\"ap-southeast-2\",\n",
+    "    arn=\"arn:aws:s3tables:ap-southeast-2:704910415367:bucket/data-index\",\n",
+    ")\n",
+    "\n",
+    "structured_table = IcebergTableConfig(\n",
+    "    catalog_config=data_index_catalog_config,\n",
+    "    namespace=\"data_index\",\n",
+    "    table_name=f\"structured_metadata_v5\",\n",
+    ").load()\n",
+    "\n",
+    "unstructured_table = IcebergTableConfig(\n",
+    "    catalog_config=data_index_catalog_config,\n",
+    "    namespace=\"data_index\",\n",
+    "    table_name=\"unstructured_metadata_v4\",\n",
+    ").load()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "f17e5c96",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# Read into memory as polars.DataFrame\n",
+    "sdf = structured_table.scan().to_polars()\n",
+    "udf = unstructured_table.scan().to_polars()\n",
+    "display(sdf)\n",
+    "display(udf)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "f97682cc",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "with polars.Config(tbl_rows=11):\n",
+    "    display(sdf[\"facility\"].value_counts(sort=True))"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "3dd3093c",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "sdf.sample(1000).write_parquet(\"small_batch.parquet\", compression=\"zstd\")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "f3cf4ee1",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# Construct generator\n",
+    "metadata = (orjson.loads(metadata) for metadata in udf[\"metadata\"])\n",
+    "\n",
+    "# Build schema\n",
+    "schema_builder = genson.SchemaBuilder()\n",
+    "for metadata in metadata:\n",
+    "    schema_builder.add_object(metadata)\n",
+    "\n",
+    "# Extract schema\n",
+    "schema = schema_builder.to_schema()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "63af009f",
+   "metadata": {},
+   "outputs": [],
+   "source": []
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "897a4c30",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# Sort global attribute keys naturally (human-like)\n",
+    "sorted_global_attribute_keys = natsort.natsorted(\n",
+    "    seq=[k for k in schema[\"properties\"][\"global_attrs\"][\"properties\"]],\n",
+    "    alg=natsort.IGNORECASE,\n",
+    ")\n",
+    "print(sorted_global_attribute_keys)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "992de4fc",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# Tuning parameters\n",
+    "SIMILARITY_THRESHOLD = 80\n",
+    "# Using token_sort_ratio handles underscores and word order swaps well\n",
+    "SCORER = thefuzz.fuzz.token_sort_ratio \n",
+    "\n",
+    "clusters = []\n",
+    "visited = set()\n",
+    "\n",
+    "for key in sorted_global_attribute_keys:\n",
+    "    if key in visited:\n",
+    "        continue\n",
+    "    \n",
+    "    # Find all items in the list that match the current key above our threshold\n",
+    "    matches = thefuzz.process.extractBests(\n",
+    "        query=key, \n",
+    "        choices=sorted_global_attribute_keys, \n",
+    "        scorer=SCORER, \n",
+    "        score_cutoff=SIMILARITY_THRESHOLD\n",
+    "    )\n",
+    "    \n",
+    "    # Filter matches to only include keys we haven't clustered yet\n",
+    "    current_cluster = [match[0] for match in matches if match[0] not in visited]\n",
+    "    \n",
+    "    if current_cluster:\n",
+    "        clusters.append(current_cluster)\n",
+    "        # Mark all keys in this cluster as visited so they aren't processed again\n",
+    "        visited.update(current_cluster)\n",
+    "\n",
+    "# Print the resulting list of lists\n",
+    "import pprint\n",
+    "pprint.pprint(clusters)"
+   ]
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "data-index (3.12.13)",
+   "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.13"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}

prefect.yaml

@@ -12,7 +12,7 @@ pull:
       repository: https://github.com/aodn/data-index.git
 
       # Set the release as the branch
-      branch: v0.4.3-rc2
+      branch: main
 
 deployments:
 

src/data_index/inventory_source/__init__.py

@@ -1,3 +1,6 @@
+from .delta_iceberg_table import (
+    DeltaIcebergTableInventorySource,
+)
 from .iceberg_table import (
     IcebergTableFacilitySubsetInventorySource,
     IcebergTableInventorySource,
@@ -6,4 +9,5 @@
 __all__ = [
     "IcebergTableInventorySource",
     "IcebergTableFacilitySubsetInventorySource",
+    "DeltaIcebergTableInventorySource",
 ]