dashsynthetic
DashSynthetic — Synthetic data generation for Databricks.
New API (profile-driven): from dashsynthetic import Profiler, Generator, OntologyConfig, ProfileStore
Legacy API (kept for backward compat): from dashsynthetic import SyntheticGenerator, MultiTableGenerator, RelationshipGraph
Launch the notebook UI:
import dashsynthetic; dashsynthetic.launch()
1""" 2DashSynthetic — Synthetic data generation for Databricks. 3 4New API (profile-driven): 5 from dashsynthetic import Profiler, Generator, OntologyConfig, ProfileStore 6 7Legacy API (kept for backward compat): 8 from dashsynthetic import SyntheticGenerator, MultiTableGenerator, RelationshipGraph 9 10Launch the notebook UI: 11 import dashsynthetic; dashsynthetic.launch() 12""" 13# ── New API ─────────────────────────────────────────────────────────────────── 14from dashsynthetic.profiler import Profiler, build_column_profile, should_reprofile 15from dashsynthetic.generator import Generator, generate_table, sample_column 16from dashsynthetic.ontology import OntologyConfig, PkFkRelationship, ReferenceRelationship 17from dashsynthetic.storage import ProfileStore 18 19# ── Legacy API ──────────────────────────────────────────────────────────────── 20from dashsynthetic.generator import SyntheticGenerator, MultiTableGenerator 21from dashsynthetic.relationships import RelationshipGraph 22 23# ── UI ──────────────────────────────────────────────────────────────────────── 24from dashsynthetic.ui import env_setup, launch 25 26__version__ = "0.1.7" 27__all__ = [ 28 # new 29 "Profiler", "Generator", "OntologyConfig", "ProfileStore", 30 "PkFkRelationship", "ReferenceRelationship", 31 "build_column_profile", "should_reprofile", 32 "generate_table", "sample_column", 33 # legacy 34 "SyntheticGenerator", "MultiTableGenerator", "RelationshipGraph", 35 # ui 36 "env_setup", "launch", 37]
276class Profiler: 277 """ 278 Profiles tables within a catalog/schema/table scope and persists the 279 results to a ProfileStore. 280 281 Usage:: 282 p = Profiler(base_path="/dbfs/dashsynthetic") 283 p.run("my_catalog") # all schemas + tables 284 p.run("my_catalog", schema="gold") # one schema 285 p.run("my_catalog", schema="gold", table="orders") # one table 286 287 Each call returns a list of result dicts:: 288 [{"table": "...", "status": "profiled"|"skipped"|"error", 289 "message": "...", "path": "..."}] 290 """ 291 292 def __init__(self, base_path: str = "/tmp/dashsynthetic"): 293 self.store = ProfileStore(base_path) 294 295 def run( 296 self, 297 catalog: str, 298 schema: Optional[str] = None, 299 table: Optional[str] = None, 300 on_progress=None, 301 ) -> list[dict]: 302 tables = self._resolve_scope(catalog, schema, table) 303 results = [] 304 for full_name in tables: 305 result = self._profile_one(full_name) 306 results.append(result) 307 if on_progress: 308 on_progress(result) 309 return results 310 311 def _resolve_scope( 312 self, 313 catalog: str, 314 schema: Optional[str], 315 table: Optional[str], 316 ) -> list[str]: 317 from pyspark.sql import SparkSession 318 spark = SparkSession.getActiveSession() 319 if spark is None: 320 raise RuntimeError("No active Spark session") 321 322 if schema and table: 323 return [f"{catalog}.{schema}.{table}"] 324 325 if schema: 326 rows = spark.sql(f"SHOW TABLES IN `{catalog}`.`{schema}`").collect() 327 return [f"{catalog}.{schema}.{r['tableName']}" for r in rows] 328 329 # Whole catalog 330 schema_rows = spark.sql(f"SHOW DATABASES IN `{catalog}`").collect() 331 all_tables: list[str] = [] 332 for sr in schema_rows: 333 sc = sr["databaseName"] 334 try: 335 trows = spark.sql(f"SHOW TABLES IN `{catalog}`.`{sc}`").collect() 336 all_tables += [f"{catalog}.{sc}.{r['tableName']}" for r in trows] 337 except Exception: 338 pass 339 return all_tables 340 341 def _profile_one(self, full_name: str) -> dict: 342 parts = full_name.strip().split(".") 343 catalog, schema, table = parts[0], parts[1], parts[2] 344 try: 345 existing = self.store.load_profile(catalog, schema, table) 346 if existing: 347 from pyspark.sql import SparkSession 348 current_count = SparkSession.getActiveSession().table(full_name).count() 349 if not should_reprofile(existing, current_count): 350 now = datetime.datetime.utcnow().isoformat() + "Z" 351 self.store.update_last_checked(catalog, schema, table, now) 352 return { 353 "table": full_name, 354 "status": "skipped", 355 "message": ( 356 f"Row count {current_count:,} — growth < 50% " 357 f"(was {existing['row_count']:,}), skipped full reprofile" 358 ), 359 "path": None, 360 } 361 362 profile = profile_table(full_name) 363 path = self.store.save_profile(profile) 364 return { 365 "table": full_name, 366 "status": "profiled", 367 "message": f"{profile['row_count']:,} rows · {profile['column_count']} columns", 368 "path": str(path), 369 } 370 371 except Exception as exc: 372 return { 373 "table": full_name, 374 "status": "error", 375 "message": str(exc), 376 "path": None, 377 }
Profiles tables within a catalog/schema/table scope and persists the results to a ProfileStore.
Usage:: p = Profiler(base_path="/dbfs/dashsynthetic") p.run("my_catalog") # all schemas + tables p.run("my_catalog", schema="gold") # one schema p.run("my_catalog", schema="gold", table="orders") # one table
Each call returns a list of result dicts:: [{"table": "...", "status": "profiled"|"skipped"|"error", "message": "...", "path": "..."}]
295 def run( 296 self, 297 catalog: str, 298 schema: Optional[str] = None, 299 table: Optional[str] = None, 300 on_progress=None, 301 ) -> list[dict]: 302 tables = self._resolve_scope(catalog, schema, table) 303 results = [] 304 for full_name in tables: 305 result = self._profile_one(full_name) 306 results.append(result) 307 if on_progress: 308 on_progress(result) 309 return results
171class Generator: 172 """ 173 Generates synthetic Delta tables for a given catalog/schema/table scope. 174 175 Reads profile JSONs from a ProfileStore, resolves PK/FK/reference 176 relationships from the OntologyConfig (if present), and writes results 177 to <output_schema>.<table>_synthetic. 178 179 Usage:: 180 g = Generator(base_path="/dbfs/dashsynthetic") 181 g.run("my_catalog") 182 g.run("my_catalog", schema="gold", output_schema="my_catalog.synthetic") 183 """ 184 185 def __init__(self, base_path: str = "/tmp/dashsynthetic"): 186 self.store = ProfileStore(base_path) 187 188 def run( 189 self, 190 catalog: str, 191 schema: Optional[str] = None, 192 table: Optional[str] = None, 193 output_schema: str = "", 194 on_progress=None, 195 ) -> list[dict]: 196 """ 197 Generate synthetic data for the given scope. 198 Returns list of result dicts: {table, status, rows, output_table, message}. 199 """ 200 profiles = self._load_profiles(catalog, schema, table) 201 if not profiles: 202 label = ".".join(filter(None, [catalog, schema, table])) 203 return [{ 204 "table": label, 205 "status": "error", 206 "rows": 0, 207 "output_table": "", 208 "message": "No profile found — run the Profiler first.", 209 }] 210 211 ontology = load_ontology(catalog, self.store) 212 table_names = [p["full_name"] for p in profiles] 213 profile_map = {p["full_name"]: p for p in profiles} 214 215 if ontology: 216 try: 217 ordered = ontology.generation_order(table_names) 218 except ValueError: 219 ordered = table_names 220 else: 221 ordered = table_names 222 223 pk_pools: dict[str, dict[str, list]] = {} # full_name → {col: [values]} 224 results = [] 225 226 for full_name in ordered: 227 if full_name not in profile_map: 228 continue 229 result = self._generate_one( 230 full_name, profile_map[full_name], ontology, pk_pools, output_schema 231 ) 232 pk_pools.setdefault(full_name, {}).update(result.pop("_pk_pools", {})) 233 results.append(result) 234 if on_progress: 235 on_progress(result) 236 237 return results 238 239 def _load_profiles( 240 self, catalog: str, schema: Optional[str], table: Optional[str] 241 ) -> list[dict]: 242 if table and schema: 243 p = self.store.load_profile(catalog, schema, table) 244 return [p] if p else [] 245 return self.store.list_profiles(catalog, schema) 246 247 def _generate_one( 248 self, 249 full_name: str, 250 profile: dict, 251 ontology: Optional[OntologyConfig], 252 pk_pools: dict, 253 output_schema: str, 254 ) -> dict: 255 try: 256 fk_pools: dict[str, list] = {} 257 ref_pools: dict[str, list] = {} 258 259 if ontology: 260 for rel in ontology.fk_relations_for(full_name): 261 parent_pool = pk_pools.get(rel.to_table, {}) 262 for fc, tc in zip(rel.from_columns, rel.to_columns): 263 if tc in parent_pool: 264 fk_pools[fc] = parent_pool[tc] 265 266 for rel in ontology.reference_relations_for(full_name): 267 ref_p = self.store.load_profile(*rel.to_table.split(".")) 268 if ref_p: 269 col_p = ref_p["columns"].get(rel.to_column, {}) 270 top_vals = col_p.get("top_values") 271 if top_vals: 272 ref_pools[rel.from_column] = [tv["value"] for tv in top_vals] 273 274 rows = generate_table(profile, fk_pools=fk_pools, ref_pools=ref_pools) 275 n = len(rows) 276 277 # Capture generated PK values for downstream FK resolution 278 new_pk_pools: dict[str, list] = {} 279 for col_name, col_p in profile.get("columns", {}).items(): 280 if col_p.get("is_likely_pk"): 281 new_pk_pools[col_name] = [r[col_name] for r in rows] 282 283 output_table = "" 284 if output_schema.strip() and rows: 285 tbl = full_name.split(".")[-1] 286 output_table = f"{output_schema}.{tbl}_synthetic" 287 self._write_to_spark(rows, output_table) 288 289 return { 290 "table": full_name, 291 "status": "generated", 292 "rows": n, 293 "output_table": output_table, 294 "message": f"Generated {n:,} rows" + (f" → {output_table}" if output_table else ""), 295 "_pk_pools": new_pk_pools, 296 } 297 298 except Exception as exc: 299 return { 300 "table": full_name, 301 "status": "error", 302 "rows": 0, 303 "output_table": "", 304 "message": str(exc), 305 "_pk_pools": {}, 306 } 307 308 @staticmethod 309 def _write_to_spark(rows: list[dict], output_table: str) -> None: 310 from pyspark.sql import SparkSession 311 spark = SparkSession.getActiveSession() 312 if spark is None: 313 raise RuntimeError("No active Spark session") 314 df = spark.createDataFrame(rows) 315 df.write.format("delta").mode("overwrite") \ 316 .option("overwriteSchema", "true") \ 317 .saveAsTable(output_table)
Generates synthetic Delta tables for a given catalog/schema/table scope.
Reads profile JSONs from a ProfileStore, resolves PK/FK/reference
relationships from the OntologyConfig (if present), and writes results
to Usage::
g = Generator(base_path="/dbfs/dashsynthetic")
g.run("my_catalog")
g.run("my_catalog", schema="gold", output_schema="my_catalog.synthetic") Generate synthetic data for the given scope.
Returns list of result dicts: {table, status, rows, output_table, message}. Remove all PK-FK relationships between from_table and to_table. Returns count removed. All tables that must be generated before this one (deduplicated, ordered). Return tables sorted so every table appears after all its parents.
Tables referenced in relationships but absent from Read/write/list table profiles and ontology config stored under base_path. All methods are pure Python — no Spark, no heavy imports. Write profile dict to disk; creates directories as needed. Return the profile dict, or None if it doesn't exist. Patch only the last_checked field without touching the rest of the profile. Return all saved profiles for a catalog, optionally filtered by schema.
Skips ontology.json and any unreadable files silently. Child table holds FK columns that reference parent table's PK columns. A column in from_table is a lookup value from a reference/dimension table. Pure Python — assembles a column profile from pre-computed aggregation
results. Called by profile_table() after all Spark aggregations finish. Returns True when the table needs a full reprofile.
Always reprofiles if no previous profile or previous count was zero.
Skips when growth is < REPROFILE_GROWTH_THRESHOLD (50%). Generate synthetic rows for one table from its profile (pure Python). Returns a list[dict], one dict per row. Dispatch to the right sampler for a single column profile.
Returns a Python list of n values (None = NULL). Legacy single-table generator kept for backward compatibility.
New code should use Generator instead. Legacy multi-table generator kept for backward compatibility. Defines which tables exist, their primary/master-data columns, and the
foreign keys linking them — so synthetic generation can run tables in
dependency order and keep FK values referentially valid. Usage::
graph = RelationshipGraph()
graph.add_table("Customer", table="catalog.schema.dim_customer", primary_key="customer_id")
graph.add_table("Account", table="catalog.schema.fact_account", primary_key="account_id",
master_data_columns=["currency_code"])
graph.add_foreign_key("Account", "customer_id", "Customer", "customer_id")
graph.generation_order() # -> ["Customer", "Account"] Foreign keys defined on Check referential integrity of tables/foreign keys; cycle detection. Topologically sort tables so every table is generated after the
tables its foreign keys point to. Raises ValueError on a dependency cycle. Open the environment setup panel (profile storage path etc.)._synthetic.
188 def run(
189 self,
190 catalog: str,
191 schema: Optional[str] = None,
192 table: Optional[str] = None,
193 output_schema: str = "",
194 on_progress=None,
195 ) -> list[dict]:
196 """
197 Generate synthetic data for the given scope.
198 Returns list of result dicts: {table, status, rows, output_table, message}.
199 """
200 profiles = self._load_profiles(catalog, schema, table)
201 if not profiles:
202 label = ".".join(filter(None, [catalog, schema, table]))
203 return [{
204 "table": label,
205 "status": "error",
206 "rows": 0,
207 "output_table": "",
208 "message": "No profile found — run the Profiler first.",
209 }]
210
211 ontology = load_ontology(catalog, self.store)
212 table_names = [p["full_name"] for p in profiles]
213 profile_map = {p["full_name"]: p for p in profiles}
214
215 if ontology:
216 try:
217 ordered = ontology.generation_order(table_names)
218 except ValueError:
219 ordered = table_names
220 else:
221 ordered = table_names
222
223 pk_pools: dict[str, dict[str, list]] = {} # full_name → {col: [values]}
224 results = []
225
226 for full_name in ordered:
227 if full_name not in profile_map:
228 continue
229 result = self._generate_one(
230 full_name, profile_map[full_name], ontology, pk_pools, output_schema
231 )
232 pk_pools.setdefault(full_name, {}).update(result.pop("_pk_pools", {}))
233 results.append(result)
234 if on_progress:
235 on_progress(result)
236
237 return results
29@dataclass
30class OntologyConfig:
31 catalog: str
32 pk_fk: list[PkFkRelationship] = field(default_factory=list)
33 reference: list[ReferenceRelationship] = field(default_factory=list)
34
35 # ── Mutators ──────────────────────────────────────────────────────────────
36
37 def add_pk_fk(
38 self,
39 from_table: str,
40 from_columns: list[str],
41 to_table: str,
42 to_columns: list[str],
43 ) -> "OntologyConfig":
44 if len(from_columns) != len(to_columns):
45 raise ValueError(
46 f"from_columns ({from_columns}) and to_columns ({to_columns}) "
47 "must have the same length"
48 )
49 self.pk_fk.append(
50 PkFkRelationship(from_table, list(from_columns), to_table, list(to_columns))
51 )
52 return self
53
54 def add_reference(
55 self,
56 from_table: str,
57 from_column: str,
58 to_table: str,
59 to_column: str,
60 ) -> "OntologyConfig":
61 self.reference.append(
62 ReferenceRelationship(from_table, from_column, to_table, to_column)
63 )
64 return self
65
66 def remove_pk_fk(self, from_table: str, to_table: str) -> int:
67 """Remove all PK-FK relationships between from_table and to_table. Returns count removed."""
68 before = len(self.pk_fk)
69 self.pk_fk = [
70 r for r in self.pk_fk
71 if not (r.from_table == from_table and r.to_table == to_table)
72 ]
73 return before - len(self.pk_fk)
74
75 def remove_reference(self, from_table: str, from_column: str) -> int:
76 before = len(self.reference)
77 self.reference = [
78 r for r in self.reference
79 if not (r.from_table == from_table and r.from_column == from_column)
80 ]
81 return before - len(self.reference)
82
83 # ── Queries ───────────────────────────────────────────────────────────────
84
85 def parents_of(self, full_table_name: str) -> list[str]:
86 """All tables that must be generated before this one (deduplicated, ordered)."""
87 seen: dict[str, None] = {}
88 for rel in self.pk_fk:
89 if rel.from_table == full_table_name:
90 seen[rel.to_table] = None
91 for rel in self.reference:
92 if rel.from_table == full_table_name:
93 seen[rel.to_table] = None
94 return list(seen)
95
96 def fk_relations_for(self, full_table_name: str) -> list[PkFkRelationship]:
97 return [r for r in self.pk_fk if r.from_table == full_table_name]
98
99 def reference_relations_for(self, full_table_name: str) -> list[ReferenceRelationship]:
100 return [r for r in self.reference if r.from_table == full_table_name]
101
102 # ── Topological sort ──────────────────────────────────────────────────────
103
104 def generation_order(self, tables: list[str]) -> list[str]:
105 """
106 Return tables sorted so every table appears after all its parents.
107 Tables referenced in relationships but absent from `tables` are
108 prepended automatically.
109 Raises ValueError on a circular dependency.
110 """
111 all_tables = list(tables)
112 # pull in any dependency not explicitly listed
113 for t in tables:
114 for p in self.parents_of(t):
115 if p not in all_tables:
116 all_tables.append(p)
117
118 deps: dict[str, list[str]] = {t: self.parents_of(t) for t in all_tables}
119
120 ordered: list[str] = []
121 visited: set[str] = set()
122 visiting: set[str] = set()
123
124 def visit(t: str) -> None:
125 if t in visited:
126 return
127 if t in visiting:
128 raise ValueError(f"Circular dependency detected involving table {t!r}")
129 visiting.add(t)
130 for dep in deps.get(t, []):
131 visit(dep)
132 visiting.discard(t)
133 visited.add(t)
134 ordered.append(t)
135
136 for t in all_tables:
137 visit(t)
138 return ordered
139
140 # ── Serialisation ─────────────────────────────────────────────────────────
141
142 def to_dict(self) -> dict:
143 return {
144 "catalog": self.catalog,
145 "pk_fk": [
146 {
147 "from_table": r.from_table,
148 "from_columns": r.from_columns,
149 "to_table": r.to_table,
150 "to_columns": r.to_columns,
151 }
152 for r in self.pk_fk
153 ],
154 "reference": [
155 {
156 "from_table": r.from_table,
157 "from_column": r.from_column,
158 "to_table": r.to_table,
159 "to_column": r.to_column,
160 }
161 for r in self.reference
162 ],
163 }
164
165 @classmethod
166 def from_dict(cls, d: dict) -> "OntologyConfig":
167 config = cls(catalog=d.get("catalog", ""))
168 for r in d.get("pk_fk", []):
169 config.pk_fk.append(
170 PkFkRelationship(
171 from_table=r["from_table"],
172 from_columns=r["from_columns"],
173 to_table=r["to_table"],
174 to_columns=r["to_columns"],
175 )
176 )
177 for r in d.get("reference", []):
178 config.reference.append(
179 ReferenceRelationship(
180 from_table=r["from_table"],
181 from_column=r["from_column"],
182 to_table=r["to_table"],
183 to_column=r["to_column"],
184 )
185 )
186 return config
187
188 def validate(self) -> list[str]:
189 issues = []
190 for r in self.pk_fk:
191 if len(r.from_columns) != len(r.to_columns):
192 issues.append(
193 f"PK-FK {r.from_table}→{r.to_table}: "
194 f"column counts differ ({r.from_columns} vs {r.to_columns})"
195 )
196 try:
197 all_tables = list({r.from_table for r in self.pk_fk} |
198 {r.to_table for r in self.pk_fk} |
199 {r.from_table for r in self.reference} |
200 {r.to_table for r in self.reference})
201 self.generation_order(all_tables)
202 except ValueError as e:
203 issues.append(str(e))
204 return issues
37 def add_pk_fk(
38 self,
39 from_table: str,
40 from_columns: list[str],
41 to_table: str,
42 to_columns: list[str],
43 ) -> "OntologyConfig":
44 if len(from_columns) != len(to_columns):
45 raise ValueError(
46 f"from_columns ({from_columns}) and to_columns ({to_columns}) "
47 "must have the same length"
48 )
49 self.pk_fk.append(
50 PkFkRelationship(from_table, list(from_columns), to_table, list(to_columns))
51 )
52 return self
66 def remove_pk_fk(self, from_table: str, to_table: str) -> int:
67 """Remove all PK-FK relationships between from_table and to_table. Returns count removed."""
68 before = len(self.pk_fk)
69 self.pk_fk = [
70 r for r in self.pk_fk
71 if not (r.from_table == from_table and r.to_table == to_table)
72 ]
73 return before - len(self.pk_fk)
85 def parents_of(self, full_table_name: str) -> list[str]:
86 """All tables that must be generated before this one (deduplicated, ordered)."""
87 seen: dict[str, None] = {}
88 for rel in self.pk_fk:
89 if rel.from_table == full_table_name:
90 seen[rel.to_table] = None
91 for rel in self.reference:
92 if rel.from_table == full_table_name:
93 seen[rel.to_table] = None
94 return list(seen)
104 def generation_order(self, tables: list[str]) -> list[str]:
105 """
106 Return tables sorted so every table appears after all its parents.
107 Tables referenced in relationships but absent from `tables` are
108 prepended automatically.
109 Raises ValueError on a circular dependency.
110 """
111 all_tables = list(tables)
112 # pull in any dependency not explicitly listed
113 for t in tables:
114 for p in self.parents_of(t):
115 if p not in all_tables:
116 all_tables.append(p)
117
118 deps: dict[str, list[str]] = {t: self.parents_of(t) for t in all_tables}
119
120 ordered: list[str] = []
121 visited: set[str] = set()
122 visiting: set[str] = set()
123
124 def visit(t: str) -> None:
125 if t in visited:
126 return
127 if t in visiting:
128 raise ValueError(f"Circular dependency detected involving table {t!r}")
129 visiting.add(t)
130 for dep in deps.get(t, []):
131 visit(dep)
132 visiting.discard(t)
133 visited.add(t)
134 ordered.append(t)
135
136 for t in all_tables:
137 visit(t)
138 return ordered
tables are
prepended automatically.
Raises ValueError on a circular dependency.142 def to_dict(self) -> dict:
143 return {
144 "catalog": self.catalog,
145 "pk_fk": [
146 {
147 "from_table": r.from_table,
148 "from_columns": r.from_columns,
149 "to_table": r.to_table,
150 "to_columns": r.to_columns,
151 }
152 for r in self.pk_fk
153 ],
154 "reference": [
155 {
156 "from_table": r.from_table,
157 "from_column": r.from_column,
158 "to_table": r.to_table,
159 "to_column": r.to_column,
160 }
161 for r in self.reference
162 ],
163 }
165 @classmethod
166 def from_dict(cls, d: dict) -> "OntologyConfig":
167 config = cls(catalog=d.get("catalog", ""))
168 for r in d.get("pk_fk", []):
169 config.pk_fk.append(
170 PkFkRelationship(
171 from_table=r["from_table"],
172 from_columns=r["from_columns"],
173 to_table=r["to_table"],
174 to_columns=r["to_columns"],
175 )
176 )
177 for r in d.get("reference", []):
178 config.reference.append(
179 ReferenceRelationship(
180 from_table=r["from_table"],
181 from_column=r["from_column"],
182 to_table=r["to_table"],
183 to_column=r["to_column"],
184 )
185 )
186 return config
188 def validate(self) -> list[str]:
189 issues = []
190 for r in self.pk_fk:
191 if len(r.from_columns) != len(r.to_columns):
192 issues.append(
193 f"PK-FK {r.from_table}→{r.to_table}: "
194 f"column counts differ ({r.from_columns} vs {r.to_columns})"
195 )
196 try:
197 all_tables = list({r.from_table for r in self.pk_fk} |
198 {r.to_table for r in self.pk_fk} |
199 {r.from_table for r in self.reference} |
200 {r.to_table for r in self.reference})
201 self.generation_order(all_tables)
202 except ValueError as e:
203 issues.append(str(e))
204 return issues
16class ProfileStore:
17 """
18 Read/write/list table profiles and ontology config stored under base_path.
19
20 All methods are pure Python — no Spark, no heavy imports.
21 """
22
23 def __init__(self, base_path: str = "/tmp/dashsynthetic"):
24 self.base_path = Path(base_path)
25
26 # ── Paths ─────────────────────────────────────────────────────────────────
27
28 def table_path(self, catalog: str, schema: str, table: str) -> Path:
29 return self.base_path / catalog / schema / f"{table}.json"
30
31 def ontology_path(self, catalog: str) -> Path:
32 return self.base_path / catalog / "ontology.json"
33
34 def catalog_dir(self, catalog: str) -> Path:
35 return self.base_path / catalog
36
37 # ── Profile CRUD ──────────────────────────────────────────────────────────
38
39 def save_profile(self, profile: dict) -> Path:
40 """Write profile dict to disk; creates directories as needed."""
41 path = self.table_path(profile["catalog"], profile["schema"], profile["table"])
42 path.parent.mkdir(parents=True, exist_ok=True)
43 path.write_text(json.dumps(profile, indent=2, default=str))
44 return path
45
46 def load_profile(self, catalog: str, schema: str, table: str) -> Optional[dict]:
47 """Return the profile dict, or None if it doesn't exist."""
48 path = self.table_path(catalog, schema, table)
49 if not path.exists():
50 return None
51 return json.loads(path.read_text())
52
53 def profile_exists(self, catalog: str, schema: str, table: str) -> bool:
54 return self.table_path(catalog, schema, table).exists()
55
56 def update_last_checked(self, catalog: str, schema: str, table: str, timestamp: str) -> bool:
57 """Patch only the last_checked field without touching the rest of the profile."""
58 path = self.table_path(catalog, schema, table)
59 if not path.exists():
60 return False
61 profile = json.loads(path.read_text())
62 profile["last_checked"] = timestamp
63 path.write_text(json.dumps(profile, indent=2, default=str))
64 return True
65
66 def list_profiles(
67 self,
68 catalog: str,
69 schema: Optional[str] = None,
70 ) -> list[dict]:
71 """
72 Return all saved profiles for a catalog, optionally filtered by schema.
73 Skips ontology.json and any unreadable files silently.
74 """
75 catalog_dir = self.catalog_dir(catalog)
76 if not catalog_dir.exists():
77 return []
78 profiles = []
79 for json_path in sorted(catalog_dir.rglob("*.json")):
80 if json_path.name == "ontology.json":
81 continue
82 if schema and json_path.parent.name != schema:
83 continue
84 try:
85 profiles.append(json.loads(json_path.read_text()))
86 except Exception:
87 pass
88 return profiles
89
90 def list_catalogs(self) -> list[str]:
91 if not self.base_path.exists():
92 return []
93 return [d.name for d in sorted(self.base_path.iterdir()) if d.is_dir()]
94
95 def list_schemas(self, catalog: str) -> list[str]:
96 catalog_dir = self.catalog_dir(catalog)
97 if not catalog_dir.exists():
98 return []
99 return [
100 d.name for d in sorted(catalog_dir.iterdir())
101 if d.is_dir()
102 ]
103
104 # ── Ontology CRUD ─────────────────────────────────────────────────────────
105
106 def save_ontology(self, catalog: str, ontology: dict) -> Path:
107 path = self.ontology_path(catalog)
108 path.parent.mkdir(parents=True, exist_ok=True)
109 path.write_text(json.dumps(ontology, indent=2, default=str))
110 return path
111
112 def load_ontology(self, catalog: str) -> Optional[dict]:
113 path = self.ontology_path(catalog)
114 if not path.exists():
115 return None
116 return json.loads(path.read_text())
39 def save_profile(self, profile: dict) -> Path:
40 """Write profile dict to disk; creates directories as needed."""
41 path = self.table_path(profile["catalog"], profile["schema"], profile["table"])
42 path.parent.mkdir(parents=True, exist_ok=True)
43 path.write_text(json.dumps(profile, indent=2, default=str))
44 return path
46 def load_profile(self, catalog: str, schema: str, table: str) -> Optional[dict]:
47 """Return the profile dict, or None if it doesn't exist."""
48 path = self.table_path(catalog, schema, table)
49 if not path.exists():
50 return None
51 return json.loads(path.read_text())
56 def update_last_checked(self, catalog: str, schema: str, table: str, timestamp: str) -> bool:
57 """Patch only the last_checked field without touching the rest of the profile."""
58 path = self.table_path(catalog, schema, table)
59 if not path.exists():
60 return False
61 profile = json.loads(path.read_text())
62 profile["last_checked"] = timestamp
63 path.write_text(json.dumps(profile, indent=2, default=str))
64 return True
66 def list_profiles(
67 self,
68 catalog: str,
69 schema: Optional[str] = None,
70 ) -> list[dict]:
71 """
72 Return all saved profiles for a catalog, optionally filtered by schema.
73 Skips ontology.json and any unreadable files silently.
74 """
75 catalog_dir = self.catalog_dir(catalog)
76 if not catalog_dir.exists():
77 return []
78 profiles = []
79 for json_path in sorted(catalog_dir.rglob("*.json")):
80 if json_path.name == "ontology.json":
81 continue
82 if schema and json_path.parent.name != schema:
83 continue
84 try:
85 profiles.append(json.loads(json_path.read_text()))
86 except Exception:
87 pass
88 return profiles
11@dataclass
12class PkFkRelationship:
13 """Child table holds FK columns that reference parent table's PK columns."""
14 from_table: str # child (holds FK)
15 from_columns: list[str] # FK columns on child
16 to_table: str # parent (holds PK)
17 to_columns: list[str] # PK columns on parent (same length as from_columns)
20@dataclass
21class ReferenceRelationship:
22 """A column in from_table is a lookup value from a reference/dimension table."""
23 from_table: str
24 from_column: str
25 to_table: str
26 to_column: str
62def build_column_profile(
63 col_name: str,
64 dtype_str: str,
65 nullable: bool,
66 null_count: int,
67 empty_count: int,
68 distinct_count: int,
69 row_count: int,
70 numeric_stats: Optional[dict] = None,
71 top_values: Optional[list[dict]] = None,
72) -> dict:
73 """
74 Pure Python — assembles a column profile from pre-computed aggregation
75 results. Called by profile_table() after all Spark aggregations finish.
76
77 Args:
78 col_name: column name (informational only)
79 dtype_str: Spark DataType string, e.g. "LongType", "StringType"
80 nullable: whether the column is nullable in the schema
81 null_count: number of NULL rows
82 empty_count: number of empty-string rows (0 for non-string columns)
83 distinct_count: approx distinct value count
84 row_count: total rows in the table
85 numeric_stats: dict with keys min/max/mean/std/p05/p25/p50/p75/p95
86 top_values: list of {"value": v, "count": c, "frequency": f}
87 """
88 null_rate = null_count / row_count if row_count > 0 else 0.0
89 empty_rate = empty_count / row_count if row_count > 0 else 0.0
90 unique_rate = distinct_count / row_count if row_count > 0 else 0.0
91
92 is_categorical = (
93 _is_boolean(dtype_str)
94 or (
95 _is_string(dtype_str)
96 and distinct_count > 0
97 and (distinct_count <= _CATEGORICAL_MAX_DISTINCT or unique_rate <= _CATEGORICAL_MAX_RATIO)
98 )
99 )
100 is_likely_pk = unique_rate >= 0.99 and null_rate == 0.0 and row_count > 10
101
102 profile: dict = {
103 "dtype": dtype_str,
104 "nullable": nullable,
105 "null_count": null_count,
106 "null_rate": round(null_rate, 6),
107 "empty_count": empty_count,
108 "empty_rate": round(empty_rate, 6),
109 "distinct_count": distinct_count,
110 "unique_rate": round(unique_rate, 6),
111 "is_categorical": is_categorical,
112 "is_likely_pk": is_likely_pk,
113 }
114
115 if numeric_stats:
116 profile["numeric_stats"] = {
117 k: (round(v, 6) if isinstance(v, float) else v)
118 for k, v in numeric_stats.items()
119 if v is not None
120 }
121
122 if top_values:
123 profile["top_values"] = top_values
124
125 return profile
Arguments:
50def should_reprofile(existing_profile: dict, current_row_count: int) -> bool:
51 """
52 Returns True when the table needs a full reprofile.
53 Always reprofiles if no previous profile or previous count was zero.
54 Skips when growth is < REPROFILE_GROWTH_THRESHOLD (50%).
55 """
56 old = existing_profile.get("row_count", 0)
57 if old == 0:
58 return True
59 return (current_row_count - old) / old >= _REPROFILE_GROWTH_THRESHOLD
122def generate_table(
123 profile: dict,
124 n_rows: Optional[int] = None,
125 fk_pools: Optional[dict] = None,
126 ref_pools: Optional[dict] = None,
127 seed: int = 42,
128) -> list[dict]:
129 """
130 Generate synthetic rows for one table from its profile (pure Python).
131
132 Args:
133 profile: table profile dict (as written by Profiler)
134 n_rows: target row count; defaults to profile["row_count"]
135 fk_pools: {col_name: [values]} — FK columns sample from these lists
136 ref_pools: {col_name: [values]} — reference columns sample from these lists
137 seed: random seed for reproducibility
138
139 Returns a list[dict], one dict per row.
140 """
141 rng = random.Random(seed)
142 n = n_rows if n_rows is not None else profile.get("row_count", 1000)
143 columns = profile.get("columns", {})
144 fk_pools = fk_pools or {}
145 ref_pools = ref_pools or {}
146
147 col_data: dict[str, list] = {}
148 for col_name, col_profile in columns.items():
149 dtype = col_profile.get("dtype", "")
150 null_rate = col_profile.get("null_rate", 0.0)
151
152 if col_name in fk_pools:
153 pool = fk_pools[col_name]
154 values = rng.choices(pool, k=n)
155 col_data[col_name] = _apply_nulls(values, null_rate, rng)
156 elif col_name in ref_pools:
157 pool = ref_pools[col_name]
158 values = rng.choices(pool, k=n)
159 col_data[col_name] = _apply_nulls(values, null_rate, rng)
160 elif col_profile.get("is_likely_pk"):
161 col_data[col_name] = generate_pk_values(n, dtype)
162 else:
163 col_data[col_name] = sample_column(col_profile, n, rng)
164
165 col_names = list(columns.keys())
166 return [{c: col_data[c][i] for c in col_names} for i in range(n)]
Arguments:
99def sample_column(col_profile: dict, n: int, rng: random.Random) -> list:
100 """
101 Dispatch to the right sampler for a single column profile.
102 Returns a Python list of n values (None = NULL).
103 """
104 dtype = col_profile.get("dtype", "")
105 null_rate = col_profile.get("null_rate", 0.0)
106 top_values = col_profile.get("top_values")
107 numeric_stats = col_profile.get("numeric_stats")
108
109 if top_values:
110 return _sample_categorical(top_values, n, null_rate, rng)
111 if _is_numeric(dtype) and numeric_stats:
112 return _sample_numeric(dtype, numeric_stats, n, null_rate, rng)
113 if _is_boolean(dtype):
114 return _sample_boolean(n, null_rate, rng)
115 if _is_timestamp(dtype):
116 return _sample_timestamp(n, null_rate, rng)
117 return _sample_string(col_profile.get("distinct_count", 100), n, null_rate, rng)
322class SyntheticGenerator:
323 """
324 Legacy single-table generator kept for backward compatibility.
325 New code should use Generator instead.
326 """
327
328 def __init__(self, df=None, table: str = None, query: str = None):
329 self._source_df = self._resolve(df, table, query)
330 self._volume: int = 0
331 self._preserve_corr: bool = True
332 self._preserve_nulls: bool = True
333 self._preserve_distributions: bool = True
334 self._output_table: Optional[str] = None
335
336 def _resolve(self, df, table, query):
337 if df is not None:
338 return df
339 try:
340 from pyspark.sql import SparkSession
341 spark = SparkSession.getActiveSession()
342 if table:
343 return spark.table(table)
344 if query:
345 return spark.sql(query)
346 except Exception as e:
347 raise ValueError(f"Could not load source: {e}")
348 raise ValueError("Provide df, table, or query")
349
350 def set_volume(self, n_rows: int):
351 self._volume = n_rows
352 return self
353
354 def preserve_correlations(self, enabled: bool = True):
355 self._preserve_corr = enabled
356 return self
357
358 def preserve_null_patterns(self, enabled: bool = True):
359 self._preserve_nulls = enabled
360 return self
361
362 def preserve_distributions(self, enabled: bool = True):
363 self._preserve_distributions = enabled
364 return self
365
366 def output_to(self, table: str):
367 self._output_table = table
368 return self
369
370 def profile(self) -> dict:
371 from dashsynthetic.profiler import profile_df
372 return profile_df(self._source_df)
373
374 def run(self):
375 from dashsynthetic.engine import generate
376 syn_df = generate(
377 source_df=self._source_df,
378 n_rows=self._volume or self._source_df.count(),
379 preserve_corr=self._preserve_corr,
380 preserve_nulls=self._preserve_nulls,
381 preserve_distributions=self._preserve_distributions,
382 )
383 if self._output_table:
384 syn_df.write.format("delta").mode("overwrite") \
385 .option("overwriteSchema", "true") \
386 .saveAsTable(self._output_table)
387 return syn_df
328 def __init__(self, df=None, table: str = None, query: str = None):
329 self._source_df = self._resolve(df, table, query)
330 self._volume: int = 0
331 self._preserve_corr: bool = True
332 self._preserve_nulls: bool = True
333 self._preserve_distributions: bool = True
334 self._output_table: Optional[str] = None
374 def run(self):
375 from dashsynthetic.engine import generate
376 syn_df = generate(
377 source_df=self._source_df,
378 n_rows=self._volume or self._source_df.count(),
379 preserve_corr=self._preserve_corr,
380 preserve_nulls=self._preserve_nulls,
381 preserve_distributions=self._preserve_distributions,
382 )
383 if self._output_table:
384 syn_df.write.format("delta").mode("overwrite") \
385 .option("overwriteSchema", "true") \
386 .saveAsTable(self._output_table)
387 return syn_df
390class MultiTableGenerator:
391 """Legacy multi-table generator kept for backward compatibility."""
392
393 def __init__(self, graph):
394 self._graph = graph
395 self._specs: dict = {}
396
397 def configure_table(self, name: str, n_rows: int = 0, preserve_corr: bool = True,
398 preserve_nulls: bool = True, preserve_distributions: bool = True,
399 output_table: Optional[str] = None):
400 from dashsynthetic.multi_engine import TableGenSpec
401 self._specs[name] = TableGenSpec(
402 n_rows=n_rows, preserve_corr=preserve_corr, preserve_nulls=preserve_nulls,
403 preserve_distributions=preserve_distributions, output_table=output_table,
404 )
405 return self
406
407 def validate(self) -> list:
408 return self._graph.validate()
409
410 def generation_order(self) -> list:
411 return self._graph.generation_order()
412
413 def run(self) -> dict:
414 from dashsynthetic.multi_engine import generate_multi
415 return generate_multi(self._graph, self._specs)
397 def configure_table(self, name: str, n_rows: int = 0, preserve_corr: bool = True,
398 preserve_nulls: bool = True, preserve_distributions: bool = True,
399 output_table: Optional[str] = None):
400 from dashsynthetic.multi_engine import TableGenSpec
401 self._specs[name] = TableGenSpec(
402 n_rows=n_rows, preserve_corr=preserve_corr, preserve_nulls=preserve_nulls,
403 preserve_distributions=preserve_distributions, output_table=output_table,
404 )
405 return self
26class RelationshipGraph:
27 """
28 Defines which tables exist, their primary/master-data columns, and the
29 foreign keys linking them — so synthetic generation can run tables in
30 dependency order and keep FK values referentially valid.
31
32 Usage::
33 graph = RelationshipGraph()
34 graph.add_table("Customer", table="catalog.schema.dim_customer", primary_key="customer_id")
35 graph.add_table("Account", table="catalog.schema.fact_account", primary_key="account_id",
36 master_data_columns=["currency_code"])
37 graph.add_foreign_key("Account", "customer_id", "Customer", "customer_id")
38 graph.generation_order() # -> ["Customer", "Account"]
39 """
40
41 def __init__(self):
42 self._tables: dict[str, TableNode] = {}
43 self._foreign_keys: list[ForeignKey] = []
44
45 def add_table(self, name: str, table: str, primary_key: str | None = None,
46 master_data_columns: list[str] | None = None):
47 self._tables[name] = TableNode(name, table, primary_key, master_data_columns or [])
48 return self
49
50 def add_foreign_key(self, from_table: str, from_column: str,
51 to_table: str, to_column: str):
52 self._foreign_keys.append(ForeignKey(from_table, from_column, to_table, to_column))
53 return self
54
55 @property
56 def tables(self) -> dict[str, TableNode]:
57 return self._tables
58
59 @property
60 def foreign_keys(self) -> list[ForeignKey]:
61 return self._foreign_keys
62
63 def foreign_keys_for(self, table_name: str) -> list[ForeignKey]:
64 """Foreign keys defined on `table_name` (i.e. it depends on their `to_table`)."""
65 return [fk for fk in self._foreign_keys if fk.from_table == table_name]
66
67 def validate(self) -> list[str]:
68 """Check referential integrity of tables/foreign keys; cycle detection."""
69 issues = []
70 for fk in self._foreign_keys:
71 if fk.from_table not in self._tables:
72 issues.append(f"Unknown table '{fk.from_table}' in foreign key")
73 if fk.to_table not in self._tables:
74 issues.append(f"Unknown table '{fk.to_table}' in foreign key")
75 try:
76 self.generation_order()
77 except ValueError as e:
78 issues.append(str(e))
79 return issues
80
81 def generation_order(self) -> list[str]:
82 """
83 Topologically sort tables so every table is generated after the
84 tables its foreign keys point to. Raises ValueError on a dependency cycle.
85 """
86 deps: dict[str, set[str]] = {name: set() for name in self._tables}
87 for fk in self._foreign_keys:
88 if fk.from_table in deps and fk.to_table in deps:
89 deps[fk.from_table].add(fk.to_table)
90
91 ordered: list[str] = []
92 visited: set[str] = set()
93 visiting: set[str] = set()
94
95 def visit(name: str):
96 if name in visited:
97 return
98 if name in visiting:
99 raise ValueError(f"Dependency cycle detected involving table '{name}'")
100 visiting.add(name)
101 for dep in sorted(deps[name]):
102 visit(dep)
103 visiting.discard(name)
104 visited.add(name)
105 ordered.append(name)
106
107 for name in sorted(self._tables):
108 visit(name)
109 return ordered
110
111 def to_dict(self) -> dict:
112 return {
113 "tables": {
114 name: {
115 "table": n.table,
116 "primary_key": n.primary_key,
117 "master_data_columns": n.master_data_columns,
118 }
119 for name, n in self._tables.items()
120 },
121 "foreign_keys": [
122 {
123 "from_table": fk.from_table, "from_column": fk.from_column,
124 "to_table": fk.to_table, "to_column": fk.to_column,
125 }
126 for fk in self._foreign_keys
127 ],
128 }
129
130 def to_json(self, indent: int = 2) -> str:
131 import json
132 return json.dumps(self.to_dict(), indent=indent)
133
134 def summary(self):
135 print(f"Tables: {len(self._tables)}")
136 print(f"Foreign keys: {len(self._foreign_keys)}")
137 issues = self.validate()
138 if issues:
139 print(f"{len(issues)} validation issue(s):")
140 for i in issues:
141 print(f" - {i}")
142 else:
143 print(f"Validation passed — generation order: {' → '.join(self.generation_order())}")
63 def foreign_keys_for(self, table_name: str) -> list[ForeignKey]:
64 """Foreign keys defined on `table_name` (i.e. it depends on their `to_table`)."""
65 return [fk for fk in self._foreign_keys if fk.from_table == table_name]
table_name (i.e. it depends on their to_table).67 def validate(self) -> list[str]:
68 """Check referential integrity of tables/foreign keys; cycle detection."""
69 issues = []
70 for fk in self._foreign_keys:
71 if fk.from_table not in self._tables:
72 issues.append(f"Unknown table '{fk.from_table}' in foreign key")
73 if fk.to_table not in self._tables:
74 issues.append(f"Unknown table '{fk.to_table}' in foreign key")
75 try:
76 self.generation_order()
77 except ValueError as e:
78 issues.append(str(e))
79 return issues
81 def generation_order(self) -> list[str]:
82 """
83 Topologically sort tables so every table is generated after the
84 tables its foreign keys point to. Raises ValueError on a dependency cycle.
85 """
86 deps: dict[str, set[str]] = {name: set() for name in self._tables}
87 for fk in self._foreign_keys:
88 if fk.from_table in deps and fk.to_table in deps:
89 deps[fk.from_table].add(fk.to_table)
90
91 ordered: list[str] = []
92 visited: set[str] = set()
93 visiting: set[str] = set()
94
95 def visit(name: str):
96 if name in visited:
97 return
98 if name in visiting:
99 raise ValueError(f"Dependency cycle detected involving table '{name}'")
100 visiting.add(name)
101 for dep in sorted(deps[name]):
102 visit(dep)
103 visiting.discard(name)
104 visited.add(name)
105 ordered.append(name)
106
107 for name in sorted(self._tables):
108 visit(name)
109 return ordered
111 def to_dict(self) -> dict:
112 return {
113 "tables": {
114 name: {
115 "table": n.table,
116 "primary_key": n.primary_key,
117 "master_data_columns": n.master_data_columns,
118 }
119 for name, n in self._tables.items()
120 },
121 "foreign_keys": [
122 {
123 "from_table": fk.from_table, "from_column": fk.from_column,
124 "to_table": fk.to_table, "to_column": fk.to_column,
125 }
126 for fk in self._foreign_keys
127 ],
128 }
134 def summary(self):
135 print(f"Tables: {len(self._tables)}")
136 print(f"Foreign keys: {len(self._foreign_keys)}")
137 issues = self.validate()
138 if issues:
139 print(f"{len(issues)} validation issue(s):")
140 for i in issues:
141 print(f" - {i}")
142 else:
143 print(f"Validation passed — generation order: {' → '.join(self.generation_order())}")
14def env_setup() -> None:
15 """Open the environment setup panel (profile storage path etc.)."""
16 try:
17 import dashui
18 from IPython.display import display
19 except ImportError:
20 raise RuntimeError("ipywidgets required. Run: %pip install ipywidgets") from None
21
22 display(dashui.card([
23 dashui.header("DashSynthetic — Environment Setup", library=_LIBRARY),
24 dashui.env_setup_panel(_LIBRARY).widget,
25 ]))
28def launch() -> None:
29 try:
30 import ipywidgets as w
31 from IPython.display import display
32 import dashui
33 except ImportError:
34 raise RuntimeError("ipywidgets required. Run: %pip install ipywidgets")
35
36 base_path = _get_base_path()
37
38 tab = w.Tab(children=[
39 _build_profile_tab(w, dashui, base_path),
40 _build_configure_tab(w, dashui, base_path),
41 _build_generate_tab(w, dashui, base_path),
42 ])
43 tab.set_title(0, "Profile")
44 tab.set_title(1, "Configure")
45 tab.set_title(2, "Generate")
46
47 env_accordion = w.Accordion(children=[dashui.env_setup_panel(_LIBRARY).widget])
48 env_accordion.set_title(0, "Environment setup")
49 env_accordion.selected_index = None
50
51 display(dashui.card([
52 dashui.header("DashSynthetic — Synthetic Data Generation", library=_LIBRARY),
53 env_accordion,
54 tab,
55 ]))