From 62b416d210e49d8cbc43c260315dd633078841df Mon Sep 17 00:00:00 2001 From: mo Date: Fri, 26 Jun 2026 01:07:23 +0000 Subject: [PATCH] infra: mirror light per-source Airflow DAGs + generators (GEN_ROWS), fix script path --- infra/airflow/per_source_gen_dags.py | 65 +++++ .../generate_cassandra_telemetry_data.py | 91 +++++++ .../scripts/generate_mongodb_events_data.py | 90 +++++++ .../scripts/generate_mysql_employee_data.py | 107 ++++++++ .../scripts/generate_neo4j_graph_data.py | 228 ++++++++++++++++++ .../scripts/generate_postgres_sales_data.py | 109 +++++++++ 6 files changed, 690 insertions(+) create mode 100644 infra/airflow/per_source_gen_dags.py create mode 100644 infra/airflow/scripts/generate_cassandra_telemetry_data.py create mode 100644 infra/airflow/scripts/generate_mongodb_events_data.py create mode 100644 infra/airflow/scripts/generate_mysql_employee_data.py create mode 100644 infra/airflow/scripts/generate_neo4j_graph_data.py create mode 100644 infra/airflow/scripts/generate_postgres_sales_data.py diff --git a/infra/airflow/per_source_gen_dags.py b/infra/airflow/per_source_gen_dags.py new file mode 100644 index 0000000..1281b86 --- /dev/null +++ b/infra/airflow/per_source_gen_dags.py @@ -0,0 +1,65 @@ +"""Per-database light data generation DAGs. + +Each DAG runs one generator script and accepts a `rows` value via the +dag_run conf (passed by the Command Center), exported as GEN_ROWS. +Triggerable independently so every database gets its own button. +""" +from airflow import DAG +from airflow.operators.python import PythonOperator +from datetime import datetime +import os +import subprocess + +SCRIPTS_DIR = "/opt/airflow/dags/scripts" + +SOURCES = { + "postgres": "generate_postgres_sales_data.py", + "mysql": "generate_mysql_employee_data.py", + "mongodb": "generate_mongodb_events_data.py", + "cassandra": "generate_cassandra_telemetry_data.py", + "neo4j": "generate_neo4j_graph_data.py", +} + +DEFAULT_ROWS = {"neo4j": "2000"} + +default_args = {"owner": "airflow", "retries": 0} + + +def make_runner(script: str, default_rows: str): + def _run(**context): + dag_run = context.get("dag_run") + conf = (dag_run.conf if dag_run else {}) or {} + rows = str(conf.get("rows") or default_rows) + env = dict(os.environ) + env["GEN_ROWS"] = rows + print(f"Running {script} with GEN_ROWS={rows}") + result = subprocess.run( + ["python3", os.path.join(SCRIPTS_DIR, script)], + capture_output=True, text=True, env=env, + ) + if result.stdout: + print(result.stdout[-4000:]) + if result.stderr: + print("STDERR:", result.stderr[-4000:]) + if result.returncode != 0: + raise Exception(f"{script} failed with return code {result.returncode}") + return _run + + +for _src, _script in SOURCES.items(): + _dag_id = f"gen_{_src}" + _dag = DAG( + dag_id=_dag_id, + default_args=default_args, + description=f"Generate light data into {_src} (GEN_ROWS via conf.rows)", + schedule=None, + start_date=datetime(2025, 1, 1), + catchup=False, + tags=["data", "generation", _src], + ) + PythonOperator( + task_id=f"generate_{_src}", + python_callable=make_runner(_script, DEFAULT_ROWS.get(_src, "5000")), + dag=_dag, + ) + globals()[_dag_id] = _dag diff --git a/infra/airflow/scripts/generate_cassandra_telemetry_data.py b/infra/airflow/scripts/generate_cassandra_telemetry_data.py new file mode 100644 index 0000000..e61ddf3 --- /dev/null +++ b/infra/airflow/scripts/generate_cassandra_telemetry_data.py @@ -0,0 +1,91 @@ +#!/usr/bin/env python3 +""" +Script to generate fake telemetry data for Cassandra +Generates approximately 1GB of data +""" + +from cassandra.cluster import Cluster +import random +from datetime import datetime, timedelta +import uuid +import sys + +# Database connection details +DB_HOST = "10.0.21.51" +DB_PORT = "9042" +KEYSPACE = "telemetry" +TABLE_NAME = "device_metrics" + +# Data generation settings +import os +TARGET_ROWS = int(os.getenv("GEN_ROWS", "5000")) # light, configurable via GEN_ROWS +BATCH_SIZE = min(5000, max(500, TARGET_ROWS)) + +# Sample data +METRIC_TYPES = ["temperature", "humidity", "pressure", "voltage", "current"] +DEVICE_PREFIX = "device-" + +def generate_fake_device_metric(): + """Generate a single fake device metric""" + device_id = f"{DEVICE_PREFIX}{random.randint(1, 50000)}" + + # Random timestamp within the last year + days_ago = random.randint(0, 365) + metric_ts = datetime.now() - timedelta(days=days_ago, hours=random.randint(0, 23), + minutes=random.randint(0, 59)) + + metric_type = random.choice(METRIC_TYPES) + metric_value = round(random.uniform(0.0, 100.0), 4) + + # Generate a long payload field + payload = "X" * 200 + + return (device_id, metric_ts, metric_type, metric_value, payload) + +def main(): + print(f"Connecting to Cassandra at {DB_HOST}:{DB_PORT}...") + + cluster = Cluster([DB_HOST], port=DB_PORT) + session = cluster.connect() + + print(f"Generating {TARGET_ROWS} device metrics...") + print(f"Batch size: {BATCH_SIZE}") + + total_generated = 0 + batch = [] + + for i in range(TARGET_ROWS): + batch.append(generate_fake_device_metric()) + + if len(batch) >= BATCH_SIZE: + session.execute( + f""" + INSERT INTO {KEYSPACE}.{TABLE_NAME} (device_id, metric_ts, metric_type, metric_value, payload) + VALUES (%s, %s, %s, %s, %s) + """, + batch + ) + total_generated += len(batch) + batch = [] + + if total_generated % 100000 == 0: + print(f"Generated {total_generated} rows...") + + # Insert remaining rows + if batch: + session.execute( + f""" + INSERT INTO {KEYSPACE}.{TABLE_NAME} (device_id, metric_ts, metric_type, metric_value, payload) + VALUES (%s, %s, %s, %s, %s) + """, + batch + ) + total_generated += len(batch) + + session.shutdown() + cluster.shutdown() + + print(f"Completed! Generated {total_generated} device metrics.") + +if __name__ == "__main__": + main() diff --git a/infra/airflow/scripts/generate_mongodb_events_data.py b/infra/airflow/scripts/generate_mongodb_events_data.py new file mode 100644 index 0000000..2be9fa8 --- /dev/null +++ b/infra/airflow/scripts/generate_mongodb_events_data.py @@ -0,0 +1,90 @@ +#!/usr/bin/env python3 +""" +Script to generate fake event data for MongoDB +Generates approximately 1GB of data +""" + +import pymongo +import random +from datetime import datetime, timedelta +import uuid +import sys + +# Database connection details +DB_HOST = "10.0.21.51" +DB_PORT = "27017" +DB_NAME = "supplychain" +COLLECTION_NAME = "events" + +# Data generation settings +import os +TARGET_DOCUMENTS = int(os.getenv("GEN_ROWS", "5000")) # light, configurable via GEN_ROWS +BATCH_SIZE = min(5000, max(500, TARGET_DOCUMENTS)) + +# Sample data +EVENT_TYPES = ["INSERT", "UPDATE", "DELETE", "CREATE", "MODIFY"] +REGIONS = ["EU", "APAC", "LATAM", "NA", "EMEA"] +SOURCES = ["ERP", "WMS", "CRM", "SCM", "TMS"] + +def generate_fake_event(): + """Generate a single fake event""" + event_id = uuid.uuid4() + event_type = random.choice(EVENT_TYPES) + region = random.choice(REGIONS) + source = random.choice(SOURCES) + + # Random timestamp within the last year + days_ago = random.randint(0, 365) + ts = datetime.now() - timedelta(days=days_ago, hours=random.randint(0, 23), + minutes=random.randint(0, 59)) + + amount = random.uniform(100.0, 50000.0) + + # Generate a long payload field (like the existing data) + payload = "X" * 500 + + return { + "event_id": event_id, + "type": event_type, + "region": region, + "source": source, + "amount": amount, + "ts": ts, + "payload": payload + } + +def main(): + print(f"Connecting to MongoDB at {DB_HOST}:{DB_PORT}...") + + client = pymongo.MongoClient(f"mongodb://{DB_HOST}:{DB_PORT}/") + db = client[DB_NAME] + collection = db[COLLECTION_NAME] + + print(f"Generating {TARGET_DOCUMENTS} events...") + print(f"Batch size: {BATCH_SIZE}") + + total_generated = 0 + batch = [] + + for i in range(TARGET_DOCUMENTS): + batch.append(generate_fake_event()) + + if len(batch) >= BATCH_SIZE: + collection.insert_many(batch) + total_generated += len(batch) + batch = [] + + if total_generated % 100000 == 0: + print(f"Generated {total_generated} documents...") + + # Insert remaining documents + if batch: + collection.insert_many(batch) + total_generated += len(batch) + + client.close() + + print(f"Completed! Generated {total_generated} events.") + +if __name__ == "__main__": + main() diff --git a/infra/airflow/scripts/generate_mysql_employee_data.py b/infra/airflow/scripts/generate_mysql_employee_data.py new file mode 100644 index 0000000..32c1791 --- /dev/null +++ b/infra/airflow/scripts/generate_mysql_employee_data.py @@ -0,0 +1,107 @@ +#!/usr/bin/env python3 +""" +Script to generate fake employee event data for MySQL +Generates approximately 1GB of data +""" + +import mysql.connector +import random +from datetime import datetime, timedelta +import uuid +import sys + +# Database connection details +DB_HOST = "10.0.21.51" +DB_PORT = "3306" +DB_NAME = "hr" +DB_USER = "mo" +DB_PASSWORD = "Dell2026!" + +# Data generation settings +import os +TARGET_ROWS = int(os.getenv("GEN_ROWS", "5000")) # light, configurable via GEN_ROWS +BATCH_SIZE = min(10000, max(500, TARGET_ROWS)) + +# Sample data +DEPARTMENTS = ["HR", "Operations", "Sales", "Marketing", "Finance", "IT", "Engineering", "Legal"] +ROLE_NAMES = ["Analyst", "Lead", "Manager", "Consultant", "Director", "Engineer", "Specialist", "Coordinator"] +REGIONS = ["EU", "APAC", "LATAM", "NA", "EMEA"] +EVENT_TYPES = ["TRANSFER", "PROMOTION", "TERMINATION", "HIRED", "SALARY_CHANGE", "DEPARTMENT_CHANGE"] + +def generate_fake_employee_event(): + """Generate a single fake employee event""" + employee_id = random.randint(1, 100000) + department = random.choice(DEPARTMENTS) + role_name = random.choice(ROLE_NAMES) + region = random.choice(REGIONS) + event_type = random.choice(EVENT_TYPES) + + # Random timestamp within the last 2 years + days_ago = random.randint(0, 730) + event_ts = datetime.now() - timedelta(days=days_ago, hours=random.randint(0, 23), + minutes=random.randint(0, 59)) + + salary_change = round(random.uniform(1000.0, 20000.0), 2) if random.random() > 0.3 else None + + # Generate a long notes field (like the existing data) + notes = str(uuid.uuid4()) * 10 + + return (employee_id, department, role_name, region, event_type, salary_change, event_ts, notes) + +def main(): + print(f"Connecting to MySQL at {DB_HOST}:{DB_PORT}...") + + conn = mysql.connector.connect( + host=DB_HOST, + port=DB_PORT, + database=DB_NAME, + user=DB_USER, + password=DB_PASSWORD + ) + cursor = conn.cursor() + + print(f"Generating {TARGET_ROWS} employee events...") + print(f"Batch size: {BATCH_SIZE}") + + total_generated = 0 + batch = [] + + for i in range(TARGET_ROWS): + batch.append(generate_fake_employee_event()) + + if len(batch) >= BATCH_SIZE: + cursor.executemany( + """ + INSERT INTO employee_events (employee_id, department, role_name, region, + event_type, salary_change, event_ts, notes) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s) + """, + batch + ) + conn.commit() + total_generated += len(batch) + batch = [] + + if total_generated % 100000 == 0: + print(f"Generated {total_generated} rows...") + + # Insert remaining rows + if batch: + cursor.executemany( + """ + INSERT INTO employee_events (employee_id, department, role_name, region, + event_type, salary_change, event_ts, notes) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s) + """, + batch + ) + conn.commit() + total_generated += len(batch) + + cursor.close() + conn.close() + + print(f"Completed! Generated {total_generated} employee events.") + +if __name__ == "__main__": + main() diff --git a/infra/airflow/scripts/generate_neo4j_graph_data.py b/infra/airflow/scripts/generate_neo4j_graph_data.py new file mode 100644 index 0000000..ef6f7b4 --- /dev/null +++ b/infra/airflow/scripts/generate_neo4j_graph_data.py @@ -0,0 +1,228 @@ +#!/usr/bin/env python3 +""" +Script to generate fake graph data for Neo4j +Generates approximately 1GB of data with nodes and relationships +""" + +from neo4j import GraphDatabase +import random +import uuid +import sys + +# Database connection details +DB_HOST = "10.0.21.51" +DB_PORT = "7687" +DB_USER = "neo4j" +DB_PASSWORD = "testpwd" + +# Data generation settings +import os +TARGET_NODES = int(os.getenv("GEN_ROWS", "2000")) # light, configurable via GEN_ROWS +BATCH_SIZE = min(1000, max(200, TARGET_NODES)) + +# Sample data +PRODUCT_CATEGORIES = ["Electronics", "Clothing", "Food", "Furniture", "Toys", "Books"] +SUPPLIER_REGIONS = ["EU", "APAC", "LATAM", "NA", "EMEA"] +RELATIONSHIP_TYPES = ["SUPPLIES", "RELATED_TO", "COMPATIBLE_WITH", "PART_OF"] + +def generate_fake_product(): + """Generate a single fake product node""" + product_id = str(uuid.uuid4()) + name = f"Product-{random.randint(1000, 999999)}" + category = random.choice(PRODUCT_CATEGORIES) + price = round(random.uniform(10.0, 1000.0), 2) + stock = random.randint(0, 1000) + + # Generate a long description field + description = "X" * 200 + + return { + "product_id": product_id, + "name": name, + "category": category, + "price": price, + "stock": stock, + "description": description + } + +def generate_fake_supplier(): + """Generate a single fake supplier node""" + supplier_id = str(uuid.uuid4()) + name = f"Supplier-{random.randint(1000, 999999)}" + region = random.choice(SUPPLIER_REGIONS) + rating = round(random.uniform(1.0, 5.0), 1) + + # Generate a long address field + address = "X" * 150 + + return { + "supplier_id": supplier_id, + "name": name, + "region": region, + "rating": rating, + "address": address + } + +def main(): + print(f"Connecting to Neo4j at {DB_HOST}:{DB_PORT}...") + + driver = GraphDatabase.driver(f"bolt://{DB_HOST}:{DB_PORT}", + auth=(DB_USER, DB_PASSWORD)) + + with driver.session() as session: + print(f"Generating {TARGET_NODES} product nodes...") + print(f"Batch size: {BATCH_SIZE}") + + total_products = 0 + total_suppliers = 0 + product_ids = [] + + # Generate product nodes + batch = [] + for i in range(TARGET_NODES): + product = generate_fake_product() + batch.append(product) + product_ids.append(product["product_id"]) + + if len(batch) >= BATCH_SIZE: + session.run( + """ + UNWIND $batch as row + CREATE (p:Product { + product_id: row.product_id, + name: row.name, + category: row.category, + price: row.price, + stock: row.stock, + description: row.description + }) + """, + batch=batch + ) + total_products += len(batch) + batch = [] + + if total_products % 50000 == 0: + print(f"Generated {total_products} product nodes...") + + # Insert remaining products + if batch: + session.run( + """ + UNWIND $batch as row + CREATE (p:Product { + product_id: row.product_id, + name: row.name, + category: row.category, + price: row.price, + stock: row.stock, + description: row.description + }) + """, + batch=batch + ) + total_products += len(batch) + + print(f"Generated {total_products} product nodes.") + + # Generate supplier nodes (fewer than products) + print(f"Generating supplier nodes...") + target_suppliers = max(20, TARGET_NODES // 10) + batch = [] + supplier_ids = [] + + for i in range(target_suppliers): + supplier = generate_fake_supplier() + batch.append(supplier) + supplier_ids.append(supplier["supplier_id"]) + + if len(batch) >= BATCH_SIZE: + session.run( + """ + UNWIND $batch as row + CREATE (s:Supplier { + supplier_id: row.supplier_id, + name: row.name, + region: row.region, + rating: row.rating, + address: row.address + }) + """, + batch=batch + ) + total_suppliers += len(batch) + batch = [] + + if batch: + session.run( + """ + UNWIND $batch as row + CREATE (s:Supplier { + supplier_id: row.supplier_id, + name: row.name, + region: row.region, + rating: row.rating, + address: row.address + }) + """, + batch=batch + ) + total_suppliers += len(batch) + + print(f"Generated {total_suppliers} supplier nodes.") + + # Create relationships between products and suppliers + print(f"Creating relationships...") + batch = [] + total_relationships = 0 + + for product_id in product_ids: + # Each product is supplied by 1-3 random suppliers + num_suppliers = random.randint(1, 3) + for _ in range(num_suppliers): + supplier_id = random.choice(supplier_ids) + rel_type = random.choice(RELATIONSHIP_TYPES) + + batch.append({ + "product_id": product_id, + "supplier_id": supplier_id, + "rel_type": rel_type + }) + + if len(batch) >= BATCH_SIZE: + session.run( + """ + UNWIND $batch as row + MATCH (p:Product {product_id: row.product_id}) + MATCH (s:Supplier {supplier_id: row.supplier_id}) + CALL apoc.create.relationship(p, row.rel_type, {}, s) YIELD rel + RETURN rel + """, + batch=batch + ) + total_relationships += len(batch) + batch = [] + + if total_relationships % 50000 == 0: + print(f"Created {total_relationships} relationships...") + + if batch: + session.run( + """ + UNWIND $batch as row + MATCH (p:Product {product_id: row.product_id}) + MATCH (s:Supplier {supplier_id: row.supplier_id}) + CALL apoc.create.relationship(p, row.rel_type, {}, s) YIELD rel + RETURN rel + """, + batch=batch + ) + total_relationships += len(batch) + + print(f"Created {total_relationships} relationships.") + + driver.close() + print(f"Completed! Generated {total_products} products, {total_suppliers} suppliers, and {total_relationships} relationships.") + +if __name__ == "__main__": + main() diff --git a/infra/airflow/scripts/generate_postgres_sales_data.py b/infra/airflow/scripts/generate_postgres_sales_data.py new file mode 100644 index 0000000..4fba1a7 --- /dev/null +++ b/infra/airflow/scripts/generate_postgres_sales_data.py @@ -0,0 +1,109 @@ +#!/usr/bin/env python3 +""" +Script to generate fake sales order data for PostgreSQL +Generates approximately 1GB of data +""" + +import psycopg2 +import random +from datetime import datetime, timedelta +import uuid +import sys +import os + +# Database connection details +DB_HOST = "10.0.21.51" +DB_PORT = "5432" +DB_NAME = "postgres" +DB_USER = "mo" +DB_PASSWORD = "Dell2026!" + +# Data generation settings +TARGET_ROWS = int(os.getenv("GEN_ROWS", "5000")) # light, configurable via GEN_ROWS +BATCH_SIZE = min(10000, max(500, TARGET_ROWS)) + +# Sample data +REGIONS = ["EU", "APAC", "LATAM", "NA", "EMEA"] +SALES_CHANNELS = ["STORE", "ONLINE", "MOBILE", "B2B"] +CURRENCIES = ["EUR", "USD", "GBP", "JPY", "CNY"] +ORDER_STATUSES = ["SHIPPED", "PENDING", "CANCELLED", "RETURNED", "DELIVERED"] + +def generate_fake_order(): + """Generate a single fake sales order""" + customer_id = random.randint(1, 100000) + product_id = random.randint(1, 5000) + region = random.choice(REGIONS) + sales_channel = random.choice(SALES_CHANNELS) + + # Random timestamp within the last 2 years + days_ago = random.randint(0, 730) + order_ts = datetime.now() - timedelta(days=days_ago, hours=random.randint(0, 23), + minutes=random.randint(0, 59)) + + amount = round(random.uniform(10.0, 10000.0), 2) + currency = random.choice(CURRENCIES) + order_status = random.choice(ORDER_STATUSES) + + # Generate a long notes field (like the existing data) + notes = str(uuid.uuid4()) * 10 + + return (customer_id, product_id, region, sales_channel, order_ts, + amount, currency, order_status, notes) + +def main(): + print(f"Connecting to PostgreSQL at {DB_HOST}:{DB_PORT}...") + + conn = psycopg2.connect( + host=DB_HOST, + port=DB_PORT, + database=DB_NAME, + user=DB_USER, + password=DB_PASSWORD + ) + cursor = conn.cursor() + + print(f"Generating {TARGET_ROWS} sales orders...") + print(f"Batch size: {BATCH_SIZE}") + + total_generated = 0 + batch = [] + + for i in range(TARGET_ROWS): + batch.append(generate_fake_order()) + + if len(batch) >= BATCH_SIZE: + cursor.executemany( + """ + INSERT INTO sales_orders (customer_id, product_id, region, sales_channel, + order_ts, amount, currency, order_status, notes) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) + """, + batch + ) + conn.commit() + total_generated += len(batch) + batch = [] + + if total_generated % 100000 == 0: + print(f"Generated {total_generated} rows...") + + # Insert remaining rows + if batch: + cursor.executemany( + """ + INSERT INTO sales_orders (customer_id, product_id, region, sales_channel, + order_ts, amount, currency, order_status, notes) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) + """, + batch + ) + conn.commit() + total_generated += len(batch) + + cursor.close() + conn.close() + + print(f"Completed! Generated {total_generated} sales orders.") + +if __name__ == "__main__": + main()