> ## Documentation Index
> Fetch the complete documentation index at: https://www.dataframer.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Single File

> Generate high-quality synthetic single-file datasets within Databricks using DataFramer.

In this notebook we will demonstrate how the [DataFramer](https://dataframer.ai) Python [SDK](https://pypi.org/project/pydataframer) (PIP Package: `pydataframer`) can be used to generate large amounts of high quality synthetic datasets where each sample size can be arbitrarily large.

## Prerequisites

Import this notebook into your Databricks workspace and run it there. To run it in Google Colab or a local Jupyter instead, see *Running it outside Databricks* below.

1. Table access (one-time admin setup)\
   Whoever runs this notebook — a person, or a service principal when it runs as a job — needs:
   * `USE CATALOG` on the catalog
   * `USE SCHEMA` and `SELECT` on the schema it reads the sample data from
   * `USE SCHEMA`, `CREATE TABLE`, `MODIFY` and `APPLY TAG` on the schema it writes results to
   * `CAN USE` on the SQL warehouse

2. DataFramer Databricks Credentials (one-time admin setup)\
   A DataFramer company admin must configure the Databricks service principal credentials in the DataFramer web application. Navigate to **Settings > Keys**, choose the provider **Databricks**, and enter:

   * **Client ID** — the service principal application (client) ID
   * **Client Secret** — the service principal secret
   * **Workspace URL** — the workspace's address\
     (e.g. `https://dbc-xxxxx.cloud.databricks.com`)

   Once saved, every AI task of every team member runs on Databricks: `databricks/` models run as chosen, and a model chosen from another provider is moved to its Databricks equivalent. A company holds one provider credential at a time, so saving the Databricks credential replaces an Anthropic or OpenAI key. No credentials need to be passed in API calls.

3. DataFramer API key\
   Retrieve it from **Settings > Keys > Copy API Key** on the [web application](https://app.dataframer.ai). Store it in a Databricks secret scope; this example uses a scope named `dataframer` with the key `DATAFRAMER_API_KEY`. Whoever runs this notebook needs at least READ permission on that scope — including the service principal, when it runs as a job.

### Running it outside Databricks

There is no workspace identity and no secret scope, so pass the values in directly. The SQL warehouse's HTTP path is on its **Connection details** tab.

```python theme={null}
databricks_connector = DatabricksConnector(
    server_hostname="dbc-xxxxx.cloud.databricks.com",
    http_path="/sql/1.0/warehouses/xxxxxxxx",
    client_id="<service principal client ID from step 2>",
    client_secret="<service principal secret from step 2>",
)
client = Dataframer(api_key="<DataFramer API key from step 3>")
```

Keep the client secret out of any notebook you share or commit.

### Step 1: Install and Setup SDK

Install the [DataFramer SDK](https://pypi.org/project/pydataframer) and the [Databricks connector package for DataFramer](https://pypi.org/project/pydataframer-databricks).

```python theme={null}
%%capture
%pip install --upgrade pydataframer pydataframer-databricks pyyaml tenacity
```

#### Initialize the DataFramer client and the DatabricksConnector

In this step, we initialize the DataFramer client with the API key stored in the `dataframer` secret scope, and the DatabricksConnector, which reaches Unity Catalog as the identity running this notebook.

```python theme={null}
import os
from datetime import datetime
from io import BytesIO
from pathlib import Path

import dataframer
import requests
import yaml
from dataframer import Dataframer
from pydataframer_databricks import DatabricksConnector
from tenacity import retry, retry_if_result, stop_never, wait_fixed

# Uses the workspace, warehouse and identity of the notebook it runs in
databricks_connector = DatabricksConnector(scope="dataframer")

client = Dataframer(api_key=databricks_connector.get_secret("DATAFRAMER_API_KEY"))

print("✓ Dataframer client initialized successfully")
print(f"Using base URL: {client.base_url}")
print(f"Dataframer SDK version: {dataframer.__version__}")
```

#### Fetch sample data

For this demo, we use sample data available in the Databricks catalog, specifically the `samples.bakehouse.media_customer_reviews` table. To keep the example lightweight, we select only the top 25 rows, export them as a CSV file, and upload the file to DataFramer.

```python theme={null}
sample_reviews_df = databricks_connector.fetch_sample_data(num_items_to_select=25, table_name="samples.bakehouse.media_customer_reviews")
```

### Step 2: Upload data

#### Prepare CSV and upload to DataFramer

To use DataFramer, only a small sample of data is needed. To derive these samples from a table, we recommend creating a CSV file that contain the relevant rows from the table and supplying that as a `Seed` dataset to DataFramer.

```python theme={null}
# Convert the sampled Pandas DataFrame to an in-memory CSV file
csv_file = BytesIO(
    sample_reviews_df.to_csv(index=False).encode("utf-8")
)
csv_file.name = "media_customer_reviews_top_25.csv"

# Upload the CSV to Dataframer
try:
    dataset = client.dataframer.seed_datasets.create_with_files(
        name="media_customer_reviews_sample_jan27_26",
        description="Top 25 rows from Databricks samples.bakehouse.media_customer_reviews",
        dataset_type="SINGLE_FILE",
        files=[csv_file]
    )
    print(f"Upload complete ✅\nDataset ID: {dataset.id}")

except Exception as e:
    if "already exists" in str(e):
        print("Dataset already exists — using existing dataset ✅")
        dataset = next(
            d for d in client.dataframer.seed_datasets.list()
            if d.name == "media_customer_reviews_sample_jan27_26"
        )
        print(f"Dataset ID: {dataset.id}")
    else:
        raise

dataset_id = dataset.id
```

#### Retrieve Dataset Details

This API demonstrates how to retrieve a specific dataset given a dataset ID.

```python theme={null}
# Get detailed information about the dataset
dataset_info = client.dataframer.seed_datasets.retrieve(dataset_id=dataset_id)

print("📋 Dataset Information:")
print("=" * 80)
print(f"ID: {dataset_info.id}")
print(f"Name: {dataset_info.name}")
print(f"Type: {dataset_info.dataset_type}")
print(f"Description: {dataset_info.description}")
print(f"Created: {dataset_info.created_at}")
print()
print(f"📁 Contents:")
print(f"  Files: {dataset_info.file_count}")
print(f"  Folders: {dataset_info.folder_count}")
print("=" * 80)
```

### Step 3: Generate Specification via the analysis API

A [specification](/docs/concepts) (or "spec") is a detailed description that captures the structure, patterns, and requirements of your data. Think of a `spec` as a blueprint for your data generation task. DataFramer automatically generates specifications by analyzing your seed data. This cell ensures a specification exists for the dataset by reusing an existing one or generating a new one.

When using a `databricks/` model, DataFramer automatically retrieves your company's stored Databricks credentials (configured by your admin in Settings > Keys).

```python theme={null}
spec_name = f"spec_for_dataset_{dataset_id}"

# --- Check for existing spec ---
specs = client.dataframer.specs.list()

existing_spec = next(
    (s for s in specs if s.name == spec_name),
    None
)

spec_id = None

if existing_spec:
    spec_id = existing_spec.id
    print("Spec already exists — reusing existing spec ✅")
else:
    spec = client.dataframer.specs.create(
        dataset_id=dataset_id,
        name=spec_name,
        spec_generation_model_name="databricks/system.ai.claude-sonnet-5",
        extrapolate_values=True,
        generate_distributions=True,
    )
    spec_id = spec.id
    print(f"Started specification generation:")
    print(f"Spec ID: {spec_id}")

    def spec_not_ready(result):
        return result.status not in ("SUCCEEDED", "FAILED")

    @retry(wait=wait_fixed(5), retry=retry_if_result(spec_not_ready), stop=stop_never)
    def poll_spec_status(client, spec_id):
        return client.dataframer.specs.retrieve(spec_id=spec_id)

    print("Polling for spec status (this may take a minute)...")
    spec_status = poll_spec_status(client, spec_id)

    if spec_status.status == "FAILED":
        raise RuntimeError(spec_status.error or "Unknown error")

    print(f"\nSpec generated successfully! ✅")

print(f"spec_id: {spec_id}")
```

#### Review Generated Specification

This cell retrieves the latest version of the generated specification and inspects key properties inferred from the dataset, such as data property variations.

```python theme={null}
# Get the spec (latest content_yaml is returned directly)
spec = client.dataframer.specs.retrieve(spec_id=spec_id)

# Parse the configuration YAML
config = yaml.safe_load(spec.content_yaml)
spec_data = config.get("spec", config)

print("\nData property variations:")
for prop in spec_data.get("data_property_variations", []):
    print(f"  • {prop['property_name']}: {len(prop['property_values'])} values")
```

### Step 4: Update Specification (Optional)

This cell demonstrates how to programmatically update a given specification. To keep this demo simple, the update is applied only when the specification is newly created (i.e., when the latest version is `1`). If the specification has already been updated, this step is skipped. In this step, we add a new data property called `Review Detail Level` with values `'Very brief', 'Brief', 'Moderate', 'Detailed'` and expected distributions `[15, 30, 35, 20]`.

```python theme={null}
# Get the spec with version history
spec = client.dataframer.specs.retrieve(spec_id=spec_id, include_versions=True)

# To keep this demo simple, only update if this is a newly created spec (1 version)
if spec.versions and len(spec.versions) > 1:
    print("ℹ️ Specification has already been updated — skipping this step")
else:
    # Parse the current config
    current_config = yaml.safe_load(spec.content_yaml)
    spec_data = current_config.get('spec', current_config)

    # Ensure data_property_variations exists
    spec_data.setdefault('data_property_variations', [])

    # Add new data property variation
    new_property = {
        'property_name': 'Review Detail Level',
        'property_values': ['Very brief', 'Brief', 'Moderate', 'Detailed'],
        'base_distributions': {
            'Very brief': 15,
            'Brief': 30,
            'Moderate': 35,
            'Detailed': 20
        },
        'conditional_distributions': {}
    }

    spec_data['data_property_variations'].append(new_property)
    print(f"✓ Added new property: {new_property['property_name']}")

    if 'requirements' in spec_data:
        spec_data['requirements'] += (
            "\n\nGenerated reviews should vary naturally in length and level of detail, "
            "while maintaining an informal customer review tone."
        )
        print("✓ Updated requirements for review context")

    # Convert back to YAML
    new_content_yaml = yaml.dump(
        current_config,
        default_flow_style=False,
        sort_keys=False
    )

    # Update the spec (creates a new version automatically)
    updated_spec = client.dataframer.specs.update(
        spec_id=spec_id,
        content_yaml=new_content_yaml
    )

    print("✓ Spec updated successfully")
```

### Step 5: Generate New Samples

Once the spec is generated and finalized after any manual modifications, we will use this spec to generate synthetic data. Databricks credentials are retrieved automatically from your company's stored configuration.

```python theme={null}
# --- Start generation run ---
run = client.dataframer.runs.create(
    spec_id=spec_id,
    generation_model="databricks/system.ai.claude-sonnet-5",
    number_of_samples=3,

    ## Advanced configuration for outline generation
    outline_model="databricks/system.ai.claude-sonnet-5",

    # revision_types=["coherence_flow", "consistency", "distinguishability", "conformance"],
    # filtering_types=["structural", "conformance"],
    # max_revision_cycles=2,
    # revision_model="databricks/system.ai.claude-sonnet-5",
)

run_id = run.id

print("Started generation run")
print(f"Run ID: {run_id}")

def run_not_finished(result):
    return result.status not in ("SUCCEEDED", "FAILED")

@retry(wait=wait_fixed(10), retry=retry_if_result(run_not_finished), stop=stop_never)
def poll_run_status(client, run_id):
    return client.dataframer.runs.retrieve(run_id=run_id)

print("Polling for run status (this may take a couple of minutes)...")
run_status = poll_run_status(client, run_id)

if run_status.status == "FAILED":
    raise RuntimeError("Generation failed")

print(f"\nGeneration completed successfully!")
print(f"Run ID: {run_id}")
print(f"Samples completed: {run_status.samples_completed}")
```

### Step 6: Evaluate Generated Samples

While DataFramer evaluates each sample as it is generated, it also supports a [post-generation evaluation](/docs/api-reference/evaluations/create). This API shows how to evaluate the generated dataset. Read the documentation for more details.

```python theme={null}
# --- Start evaluation ---
print(f"Creating evaluation for run: {run_id}")

evaluation = client.dataframer.evaluations.create(
    run_id=run_id,
    evaluation_model="databricks/system.ai.claude-sonnet-5"
)

evaluation_id = evaluation.id

print("\nEvaluation created")
print(f"Evaluation ID: {evaluation_id}")
print(f"Created at   : {evaluation.created_at}")

def eval_not_finished(result):
    return result.status not in ("SUCCEEDED", "FAILED")

@retry(wait=wait_fixed(5), retry=retry_if_result(eval_not_finished), stop=stop_never)
def poll_eval_status(client, evaluation_id):
    return client.dataframer.evaluations.retrieve(evaluation_id=evaluation_id)

print("Polling for evaluation status...")
eval_status = poll_eval_status(client, evaluation_id)

if eval_status.status == "FAILED":
    print("\nEvaluation failed. ❌")
    if eval_status.error_message:
        print(f"  Error: {eval_status.error_message}")
else:
    print("\nEvaluation completed successfully! ✅")
```

### Step 7: Download Generated Files

#### List Generated Files

This API lists all the files that were present in the generated dataset.

```python theme={null}
# Get generated files for the run
run_details = client.dataframer.runs.retrieve(run_id=run_id)

print("📁 Generated Files:")
print("=" * 80)
print(f"Run ID: {run_id}")
if run_details.generated_files:
    print(f"Total files: {len(run_details.generated_files)}")
    print("=" * 80)

    for i, file in enumerate(run_details.generated_files, 1):
        print(f"\n📄 File {i}:")
        print(f"  Path: {file.path}")
        print(f"  ID: {file.id}")
        print(f"  Size: {file.size_bytes} bytes")
        print(f"  Type: {file.file_type}")
else:
    print("No generated files found")
    print("=" * 80)
```

#### Download All Files as ZIP

This API allows you to download all the generated files with metadata as a compressed ZIP file.

```python theme={null}
print("📥 Downloading generated files with metadata as ZIP...")

def download_not_ready(response):
    return not hasattr(response, 'download_url') or response.download_url is None

@retry(wait=wait_fixed(2), retry=retry_if_result(download_not_ready), stop=stop_never,
       before_sleep=lambda rs: print("  ZIP generation in progress, waiting..."))
def poll_download(client, run_id):
    return client.dataframer.runs.files.download_all(run_id=run_id)

response = poll_download(client, run_id)

# Download the ZIP from the presigned URL
zip_content = requests.get(response.download_url).content
downloaded_zip = BytesIO(zip_content)

print(f"\n✅ Downloaded ZIP file ({response.size_bytes} bytes)")
```

#### Load generated data into a Delta table

This cell writes the generated data into a Delta table `<catalog>.<schema>.<table_name>`. If the table already exists, its contents are overwritten.

```python theme={null}
from pydataframer_databricks import DatasetType, FileType

databricks_connector.load_to_table(table_name="workspace.default.generated_samples",
                                        downloaded_zip=downloaded_zip,
                                        dataset_type=DatasetType(dataset_info.dataset_type),
                                        file_type=FileType.CSV
                                    )
```


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.