infra: historical sales -> iceberg.historical (Trino) + Superset dashboard scripts

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")

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])

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)

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])