Databricks & Apache Spark

The misata.spark module writes Misata-generated data straight into Delta Lake and imports existing Spark schemas back into Misata. It runs on Databricks (Free Edition or full), AWS Glue, EMR, Dataproc, or any local PySpark 3.3+ environment.

Misata's value on Spark over the usual dbldatagen:

• Multiple related tables in one call: dbldatagen builds one table at a time with no awareness of relationships.
• Referential integrity guaranteed and verifiable: zero orphan foreign keys, checked in Delta with anti-joins.
• Realistic distributions: lognormal, Poisson, normal, uniform, and correlated columns.
• Outcome conformance: declare an exact aggregate or rate (e.g. a monthly fraud rate) and the generated data conforms, giving pipeline tests a known ground truth to assert against.

Installation#

Databricks serverless / Free Edition

Install plain misata: not misata[spark]. PySpark is already on every Databricks cluster, and installing it on a serverless notebook will stop the session. The module imports the cluster's PySpark lazily.

%pip install misata
dbutils.library.restartPython()

Local / Glue / EMR / self-managed Spark

pip install misata[spark]    # pulls pyspark>=3.3.0

The module never imports PySpark at module-load time, so import misata stays lightweight. Every function calls import pyspark lazily and raises a clear error if it is missing.

Quick start#

import misata
from misata import spark as mspark

schema = misata.from_dict_schema({
    "customers": {
        "__rows__": 500,
        "id":      {"type": "integer", "primary_key": True},
        "email":   {"type": "email"},
        "country": {"type": "string", "text_type": "country"},
    },
    "orders": {
        "__rows__": 2000,
        "id":          {"type": "integer", "primary_key": True},
        "customer_id": {"type": "integer", "foreign_key": {"table": "customers", "column": "id"}},
        "total":       {"type": "float", "distribution": "lognormal", "mu": 4.5, "sigma": 0.9, "min": 4.99},
    },
})

result = mspark.generate_to_delta(schema, spark, catalog="dev", database="bronze", mode="overwrite")
print(result.summary())
result.raise_on_error()

spark is the active SparkSession, pre-defined in every Databricks notebook.

API reference#

to_spark(tables, spark, *, infer_schema=False, date_columns=None, schema_config=None)#

Convert Misata's {table_name: pd.DataFrame} output to {table_name: SparkDataFrame} using an explicit, type-correct schema. Spark's own inference can widen nullable integers to doubles or reject datetime.date objects; an explicit schema avoids both.

tables = misata.generate_from_schema(schema)
spark_tables = mspark.to_spark(tables, spark, schema_config=schema)
spark_tables["orders"].printSchema()

Pass schema_config so columns declared type: "date" are mapped to Delta DateType (rather than TimestampType). Misata stores both date and datetime as pandas datetime64[ns], so the schema is the only way to recover the distinction.

write_delta(tables, spark, *, catalog=None, database=None, schema=None, mode="overwrite", partition_by=None, cluster_by=None, merge_keys=None, table_properties=None, optimize_after_write=False, create_schema_if_not_exists=True, date_columns=None, schema_config=None, verbose=True)#

Write every table to Delta Lake. Returns a DeltaWriteResult.

mspark.write_delta(
    tables, spark,
    catalog="dev", database="bronze",
    mode="overwrite",
    partition_by={"events": ["event_date"]},
    cluster_by={"transactions": ["txn_month"]},          # Delta liquid clustering
    table_properties={"delta.autoOptimize.optimizeWrite": "true"},
    optimize_after_write=True,
    schema_config=schema,                                 # preserve date typing
)

Naming. catalog, database/schema, and the table name assemble into Unity Catalog 3-part names (catalog.database.table), 2-part (database.table), or bare table names, whatever you provide.

Write modes:

# Upsert: re-running this updates matched rows instead of duplicating them
mspark.write_delta(tables, spark, database="bronze",
                   mode="merge", merge_keys={"customers": ["id"]})

cluster_by uses Delta's clusterBy writer when available and falls back gracefully (writing without clustering, with a warning) on older Delta builds.

verify_delta_integrity(spark, relationships, *, catalog=None, database=None, schema=None, sample_size=5)#

Verify referential integrity of Delta tables with a Spark SQL LEFT ANTI JOIN per relationship. Returns a SparkIntegrityReport.

relationships = [
    {"from_table": "orders",      "from_column": "customer_id", "to_table": "customers", "to_column": "id"},
    {"from_table": "order_items", "from_column": "order_id",    "to_table": "orders",    "to_column": "id"},
]
report = mspark.verify_delta_integrity(spark, relationships, catalog="dev", database="bronze")
print(report.summary())
report.raise_if_invalid()       # raises ValueError listing orphan counts + samples

Shorthand dot-notation is also accepted: {"from": "orders.customer_id", "to": "customers.id"}.

from_spark_schema(source, spark=None, *, table_name="table", row_count=1000, foreign_keys=None, seed=42)#

Convert a single Spark schema into a Misata SchemaConfig. source may be a StructType, a DataFrame, or a fully-qualified table-name string. Reads only the public .schema, so it works on Spark Connect / serverless.

schema = mspark.from_spark_schema(
    spark.table("prod.bronze.orders"),
    row_count=5000,
    foreign_keys={"customer_id": {"table": "customers", "column": "id"}},
)

from_catalog_table(table_name, spark, *, row_count=1000, foreign_keys=None, seed=42)#

Import one Unity Catalog / Hive table by name.

schema = mspark.from_catalog_table("dev.bronze.orders", spark, row_count=5000)

from_catalog_schema(spark, database, *, catalog=None, row_counts=None, foreign_keys=None, infer_foreign_keys=True, seed=42)#

Import every table in a database and assemble a multi-table SchemaConfig. FK relationships are inferred from {parent}_id column naming (de-pluralisation aware: order_id → orders.id). Unmapped *_id columns produce a warning rather than a silent miss.

schema = mspark.from_catalog_schema(
    spark, database="bronze", catalog="prod",
    row_counts={"customers": 500, "orders": 2000, "order_items": 6000},
    foreign_keys={                                   # override / supplement inference
        "order_items": {"sku_ref": {"table": "products", "column": "sku"}},
    },
)
tables = misata.generate_from_schema(schema)

Use this to mirror a production schema into a dev environment: structure only, no data copied, and generate matching synthetic data.

append_to_delta(schema_config, spark, n_rows, *, catalog=None, database=None, schema=None, seed=None, verbose=True)#

Generate additional rows and append them to existing Delta tables. Reads each table's current MAX(id) to offset new primary keys so they never collide. FK integrity is maintained within the new batch. Date typing conforms to the existing target table.

mspark.append_to_delta(schema, spark, n_rows={"customers": 200, "orders": 800}, database="bronze")

write_delta_stream(schema_config, spark, *, catalog=None, database=None, schema=None, batch_size=100_000, partition_by=None, table_properties=None, optimize_after_write=False, create_schema_if_not_exists=True, verbose=True)#

Stream-write very large datasets. Uses Misata's batch generator so the full dataset is never buffered in memory, suitable for tens or hundreds of millions of rows.

schema.tables[0].row_count = 50_000_000
mspark.write_delta_stream(schema, spark, database="bronze", batch_size=250_000)

generate_to_spark(schema_or_story, spark, *, rows=10_000, seed=None, smart_correlations=False)#

One-liner: generate and return Spark DataFrames. Accepts a SchemaConfig or a plain-English story string.

spark_tables = mspark.generate_to_spark("An ecommerce store with 5k orders", spark)

generate_to_delta(schema_or_story, spark, *, catalog=None, database=None, schema=None, rows=10_000, seed=None, mode="overwrite", partition_by=None, cluster_by=None, table_properties=None, optimize_after_write=False, create_schema_if_not_exists=True, smart_correlations=False, verbose=True)#

One-liner: generate and write to Delta. Accepts a SchemaConfig or a story string.

Result types#

DeltaWriteResult#

Returned by write_delta, write_delta_stream, append_to_delta, generate_to_delta.

SparkIntegrityReport#

Returned by verify_delta_integrity.

Each SparkIntegrityViolation carries child_table, child_column, parent_table, parent_column, orphan_count, and sample_orphan_values.

End-to-end tutorial#

examples/databricks/medallion_fraud_pipeline.py builds a complete fraud-detection medallion pipeline:

1. Bronze: generate four FK-linked tables (customers, accounts, merchants, transactions) with a declared monthly fraud-rate curve (1.8% → 4.1%).
2. Silver: join all four tables, derive fraud-signal features.
3. Gold: aggregate monthly fraud rate and dollar volume.
4. Assert the Gold output matches the declared fraud rate, a CI-grade correctness test that's impossible with Faker or dbldatagen, because they give you no ground truth to check.

Import the .py file into Databricks (Workspace → Import → File) and Run all on serverless.

Compatibility#

The full integration is covered by tests/test_spark.py, guarded by pytest.importorskip("pyspark") so it runs in CI/Databricks and skips cleanly elsewhere.