diff --git a/infra/historical/historical_postgres.sql b/infra/historical/historical_postgres.sql deleted file mode 100644 index 44eb4d1..0000000 --- a/infra/historical/historical_postgres.sql +++ /dev/null @@ -1,18 +0,0 @@ -DROP TABLE IF EXISTS public.historical_sales; -CREATE TABLE public.historical_sales ( - order_id bigint, order_date date, order_year int, region text, - product_category text, channel text, quantity int, - unit_price double precision, amount double precision -); -INSERT INTO public.historical_sales -SELECT s, - DATE '2019-01-01' + ((s*7)%2190), - EXTRACT(YEAR FROM (DATE '2019-01-01' + ((s*7)%2190)))::int, - (ARRAY['EMEA','AMER','APAC','LATAM'])[(s%4)+1], - (ARRAY['Servers','Storage','Networking','Laptops','Services'])[(s%5)+1], - (ARRAY['Direct','Partner','Online'])[(s%3)+1], - (1+(s%50))::int, - round((100+(s%9000)*0.5)::numeric,2)::double precision, - round(((1+(s%50))*(100+(s%9000)*0.5))::numeric,2)::double precision -FROM generate_series(1,25000) s; -SELECT order_year, count(*), round(sum(amount)::numeric,0) FROM public.historical_sales GROUP BY order_year ORDER BY 1; diff --git a/infra/historical/historical_s3_load.py b/infra/historical/historical_s3_load.py new file mode 100644 index 0000000..5d444b6 --- /dev/null +++ b/infra/historical/historical_s3_load.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.s3lake.historical_sales +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}:", e.get("message") if e else "ok") diff --git a/infra/historical/superset_repoint_durable.py b/infra/historical/superset_repoint_s3.py similarity index 74% rename from infra/historical/superset_repoint_durable.py rename to infra/historical/superset_repoint_s3.py index 606efa4..a72355c 100644 --- a/infra/historical/superset_repoint_durable.py +++ b/infra/historical/superset_repoint_s3.py @@ -7,18 +7,18 @@ with app.app_context(): from superset.models.core import Database t = db.session.query(SqlaTable).filter_by(table_name="historical_sales").first() - pg = ( + ice = ( db.session.query(Database) - .filter(Database.database_name.like("%PostgreSQL Sales%")) + .filter(Database.database_name.like("%Iceberg Lakehouse%")) .first() ) - t.database_id = pg.id - t.schema = "public" + t.database_id = ice.id + t.schema = "s3lake" db.session.commit() try: t.fetch_metadata() db.session.commit() except Exception as e: print("fetch_metadata warn:", str(e)[:200]) - print("repointed dataset", t.id, "-> db", pg.id, pg.database_name, + print("repointed dataset", t.id, "-> db", ice.id, ice.database_name, "schema", t.schema, "cols", [c.column_name for c in t.columns])