From f4cd33b784e19dc3bf5782e26dd1602ee41f2b84 Mon Sep 17 00:00:00 2001 From: mo Date: Fri, 26 Jun 2026 09:04:33 +0000 Subject: [PATCH] infra: historical sales -> iceberg.historical (Trino) + Superset dashboard scripts --- infra/historical/histload.py | 26 ++++++++ infra/historical/superset_dashboard.py | 82 ++++++++++++++++++++++++++ infra/historical/superset_setup.py | 42 +++++++++++++ infra/historical/tq.py | 14 +++++ 4 files changed, 164 insertions(+) create mode 100644 infra/historical/histload.py create mode 100644 infra/historical/superset_dashboard.py create mode 100644 infra/historical/superset_setup.py create mode 100644 infra/historical/tq.py diff --git a/infra/historical/histload.py b/infra/historical/histload.py new file mode 100644 index 0000000..d80606a --- /dev/null +++ b/infra/historical/histload.py @@ -0,0 +1,26 @@ +import json, urllib.request +Q="\x27" +def run(sql): + req=urllib.request.Request("http://10.0.21.50:8089/v1/statement", data=sql.encode(), headers={"X-Trino-User":"admin"}) + d=json.load(urllib.request.urlopen(req)); nu=d.get("nextUri"); err=d.get("error") + while nu: + j=json.load(urllib.request.urlopen(nu)) + if j.get("error"): err=j["error"] + nu=j.get("nextUri") + return err +def chunk(off, size): + sql=f"""INSERT INTO iceberg.historical.sales_orders +SELECT order_id, order_date, year(order_date) AS order_year, region, product_category, channel, quantity, unit_price, ROUND(quantity*unit_price,2) AS amount +FROM ( + SELECT (s+{off}) AS order_id, + DATE {Q}2019-01-01{Q} + CAST(((s+{off})*7)%2190 AS bigint) * INTERVAL {Q}1{Q} DAY AS order_date, + element_at(ARRAY[{Q}EMEA{Q},{Q}AMER{Q},{Q}APAC{Q},{Q}LATAM{Q}], CAST((s+{off})%4 AS int)+1) AS region, + element_at(ARRAY[{Q}Servers{Q},{Q}Storage{Q},{Q}Networking{Q},{Q}Laptops{Q},{Q}Services{Q}], CAST((s+{off})%5 AS int)+1) AS product_category, + element_at(ARRAY[{Q}Direct{Q},{Q}Partner{Q},{Q}Online{Q}], CAST((s+{off})%3 AS int)+1) AS channel, + CAST(1+((s+{off})%50) AS int) AS quantity, + ROUND(100+((s+{off})%9000)*0.5,2) AS unit_price + FROM UNNEST(SEQUENCE(1,{size})) AS t(s) +)""" + return run(sql) +for off,size in [(0,10000),(10000,10000),(20000,5000)]: + e=chunk(off,size); print(f"chunk off={off} size={size}:", e.get("message") if e else "ok") diff --git a/infra/historical/superset_dashboard.py b/infra/historical/superset_dashboard.py new file mode 100644 index 0000000..b0a4838 --- /dev/null +++ b/infra/historical/superset_dashboard.py @@ -0,0 +1,82 @@ +import json +from superset.app import create_app + +app = create_app() +with app.app_context(): + from superset import db + from superset.models.slice import Slice + from superset.models.dashboard import Dashboard + from superset.connectors.sqla.models import SqlaTable + + tbl = ( + db.session.query(SqlaTable) + .filter_by(table_name="historical_sales", schema="historical") + .first() + ) + ds_id = tbl.id + + rev = {"expressionType": "SIMPLE", "column": {"column_name": "amount"}, + "aggregate": "SUM", "label": "Revenue", "hasCustomLabel": True} + cnt = {"expressionType": "SIMPLE", "column": {"column_name": "order_id"}, + "aggregate": "COUNT", "label": "Orders", "hasCustomLabel": True} + + def upsert_slice(name, viz_type, params): + sl = db.session.query(Slice).filter_by(slice_name=name).first() + if not sl: + sl = Slice(slice_name=name, datasource_type="table", datasource_id=ds_id, + viz_type=viz_type, params=json.dumps(params)) + db.session.add(sl) + else: + sl.viz_type = viz_type + sl.params = json.dumps(params) + sl.datasource_id = ds_id + db.session.commit() + return sl + + s_year = upsert_slice( + "Historical Revenue by Year", "table", + {"viz_type": "table", "datasource": f"{ds_id}__table", + "query_mode": "aggregate", "groupby": ["order_year"], + "metrics": [rev, cnt], "row_limit": 100, + "order_desc": True, "adhoc_filters": []}, + ) + s_region = upsert_slice( + "Historical Revenue by Region", "pie", + {"viz_type": "pie", "datasource": f"{ds_id}__table", + "groupby": ["region"], "metric": rev, "row_limit": 100, + "adhoc_filters": []}, + ) + s_cat = upsert_slice( + "Historical Revenue by Category", "pie", + {"viz_type": "pie", "datasource": f"{ds_id}__table", + "groupby": ["product_category"], "metric": rev, "row_limit": 100, + "adhoc_filters": []}, + ) + + title = "Historical Sales (HDFS archive)" + dash = db.session.query(Dashboard).filter_by(slug="historical-sales").first() + if not dash: + dash = Dashboard(dashboard_title=title, slug="historical-sales", published=True) + db.session.add(dash) + dash.slices = [s_year, s_region, s_cat] + + def chart_node(cid, slice_obj, w, h): + return {"type": "CHART", "id": cid, "children": [], + "parents": ["ROOT_ID", "GRID_ID", "ROW-1"], + "meta": {"chartId": slice_obj.id, "width": w, "height": h, + "sliceName": slice_obj.slice_name}} + + pos = { + "DASHBOARD_VERSION_KEY": "v2", + "ROOT_ID": {"type": "ROOT", "id": "ROOT_ID", "children": ["GRID_ID"]}, + "GRID_ID": {"type": "GRID", "id": "GRID_ID", "children": ["ROW-1"], "parents": ["ROOT_ID"]}, + "HEADER_ID": {"type": "HEADER", "id": "HEADER_ID", "meta": {"text": title}}, + "ROW-1": {"type": "ROW", "id": "ROW-1", "children": ["CHART-A", "CHART-B", "CHART-C"], + "parents": ["ROOT_ID", "GRID_ID"], "meta": {"background": "BACKGROUND_TRANSPARENT"}}, + "CHART-A": chart_node("CHART-A", s_year, 4, 50), + "CHART-B": chart_node("CHART-B", s_region, 4, 50), + "CHART-C": chart_node("CHART-C", s_cat, 4, 50), + } + dash.position_json = json.dumps(pos) + db.session.commit() + print("DASHBOARD_ID", dash.id, "slug", dash.slug, "slices", [s.id for s in dash.slices]) diff --git a/infra/historical/superset_setup.py b/infra/historical/superset_setup.py new file mode 100644 index 0000000..dedef39 --- /dev/null +++ b/infra/historical/superset_setup.py @@ -0,0 +1,42 @@ +import json +from superset.app import create_app + +app = create_app() +with app.app_context(): + from superset import db + from superset.models.core import Database + from superset.connectors.sqla.models import SqlaTable + + name = "Trino \u00b7 Iceberg Lakehouse" + uri = "trino://admin@10.0.21.50:8089/iceberg" + + dbobj = db.session.query(Database).filter_by(database_name=name).first() + if not dbobj: + dbobj = Database(database_name=name, sqlalchemy_uri=uri) + dbobj.expose_in_sqllab = True + db.session.add(dbobj) + db.session.commit() + print("created database id", dbobj.id) + else: + print("database exists id", dbobj.id) + + tbl = ( + db.session.query(SqlaTable) + .filter_by(table_name="historical_sales", schema="historical", database_id=dbobj.id) + .first() + ) + if not tbl: + tbl = SqlaTable(table_name="historical_sales", schema="historical", database=dbobj) + db.session.add(tbl) + db.session.commit() + try: + tbl.fetch_metadata() + db.session.commit() + except Exception as e: + print("fetch_metadata warning:", str(e)[:200]) + print("created dataset id", tbl.id) + else: + print("dataset exists id", tbl.id) + + print("COLUMNS:", [c.column_name for c in tbl.columns]) + print("DB_ID", dbobj.id, "TABLE_ID", tbl.id) diff --git a/infra/historical/tq.py b/infra/historical/tq.py new file mode 100644 index 0000000..bf5d984 --- /dev/null +++ b/infra/historical/tq.py @@ -0,0 +1,14 @@ +import sys, json, urllib.request +def q(sql): + req=urllib.request.Request("http://10.0.21.50:8089/v1/statement", data=sql.encode(), headers={"X-Trino-User":"admin"}) + d=json.load(urllib.request.urlopen(req)) + rows=d.get("data") or []; nu=d.get("nextUri"); err=None + while nu: + j=json.load(urllib.request.urlopen(nu)); rows+=j.get("data") or []; + if j.get("error"): err=j["error"] + nu=j.get("nextUri") + return rows, err +sql=sys.argv[1] +rows,err=q(sql) +if err: print("ERROR:", err.get("message")) +else: print("OK", rows[:5])