diff --git a/infra/airflow/scripts/hadoop_history_load.py b/infra/airflow/scripts/hadoop_history_load.py new file mode 100644 index 0000000..9f1ee0f --- /dev/null +++ b/infra/airflow/scripts/hadoop_history_load.py @@ -0,0 +1,72 @@ +#!/usr/bin/env python3 +"""Generate historical sales data and load it into HDFS (partitioned by year). + +Runs on the Hadoop master node (has hdfs client + datanode access). Writes +CSV partitions under HDFS_BASE so the data is browsable in the Command Center +HDFS view and can later be exposed as a Hive/Iceberg external table in Trino. +""" +import csv +import os +import random +import subprocess +import tempfile +import uuid +from datetime import datetime, timedelta + +HDFS_BASE = os.getenv("HDFS_BASE", "/data/historical/sales_orders") +ROWS_PER_YEAR = int(os.getenv("ROWS_PER_YEAR", "5000")) +YEARS = [int(y) for y in os.getenv("YEARS", "2020,2021,2022,2023,2024").split(",")] + +REGIONS = ["EU", "APAC", "LATAM", "NA", "EMEA"] +CHANNELS = ["STORE", "ONLINE", "MOBILE", "B2B"] +CURRENCIES = ["EUR", "USD", "GBP", "JPY", "CNY"] +STATUSES = ["SHIPPED", "PENDING", "CANCELLED", "RETURNED", "DELIVERED"] +HEADER = ["order_id", "customer_id", "product_id", "region", "sales_channel", + "order_ts", "amount", "currency", "order_status"] + + +def hdfs(*args): + subprocess.run(["hdfs", "dfs", *args], check=True) + + +def gen_year(year): + start = datetime(year, 1, 1) + rows = [] + for _ in range(ROWS_PER_YEAR): + ts = start + timedelta(days=random.randint(0, 364), hours=random.randint(0, 23), + minutes=random.randint(0, 59)) + rows.append([ + str(uuid.uuid4()), + random.randint(1, 100000), + random.randint(1, 5000), + random.choice(REGIONS), + random.choice(CHANNELS), + ts.strftime("%Y-%m-%d %H:%M:%S"), + round(random.uniform(10.0, 10000.0), 2), + random.choice(CURRENCIES), + random.choice(STATUSES), + ]) + return rows + + +def main(): + print(f"Loading historical sales into hdfs://{HDFS_BASE} for years {YEARS}") + total = 0 + for year in YEARS: + rows = gen_year(year) + with tempfile.NamedTemporaryFile("w", suffix=".csv", delete=False, newline="") as f: + w = csv.writer(f) + w.writerow(HEADER) + w.writerows(rows) + local = f.name + hdfs_dir = f"{HDFS_BASE}/year={year}" + hdfs("-mkdir", "-p", hdfs_dir) + hdfs("-put", "-f", local, f"{hdfs_dir}/part-0.csv") + os.unlink(local) + total += len(rows) + print(f" year={year}: {len(rows)} rows -> {hdfs_dir}/part-0.csv") + print(f"Done. {total} historical rows across {len(YEARS)} years in HDFS.") + + +if __name__ == "__main__": + main()