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

# Workflows

> How the Data Governance solution uses Prefect for workflow orchestration, including profiling, enrichment, and DQ rule generation workflows.

# Workflow Orchestration

The Data Governance solution uses **Prefect** for orchestrating data profiling, metadata enrichment, and data quality (DQ) rule generation workflows. Prefect provides task scheduling, concurrency control, observability, and failure handling.

## Architecture

<Tabs>
  <Tab title="Infrastructure">
    | Component          | Purpose                                      |
    | ------------------ | -------------------------------------------- |
    | **Prefect Server** | Workflow API and UI (port 4200)              |
    | **Prefect Worker** | Executes flows in Kubernetes pods            |
    | **Work Pool**      | Kubernetes work pool named `kubernetes-pool` |
    | **PostgreSQL**     | Prefect metadata storage                     |

    All Prefect services are deployed alongside the Data Governance API via Docker Compose (local dev) or Helm (Kubernetes).
  </Tab>

  <Tab title="Flow Definitions">
    Flow definitions are managed as code in `scripts/prefect/deploy.py` -- the single source of truth for all deployments. This file registers flows, configures schedules, and sets job variables.

    ```python theme={null}
    # scripts/prefect/deploy.py (simplified)
    from prefect import flow
    from prefect.deployments import Deployment

    deployments = [
        Deployment.build_from_flow(
            flow=profile_database,
            name="profile-database",
            work_pool_name="kubernetes-pool",
            job_variables={...}
        ),
        Deployment.build_from_flow(
            flow=enrich_metadata,
            name="enrich-metadata",
            work_pool_name="kubernetes-pool",
            job_variables={...}
        ),
    ]
    ```

    <Note>
      `prefect.yaml` is deprecated. All flow definitions are managed in `deploy.py`. Secrets are read from pod environment variables (injected via Helm secretKeyRef) and passed as plain strings to Prefect job\_variables.
    </Note>
  </Tab>
</Tabs>

## Available Workflows

```mermaid theme={null}
%%{init: {'theme': 'base', 'themeVariables': {'lineColor': '#555555', 'fontFamily': 'sans-serif', 'edgeLabelBackground': '#ffffff'}}}%%
flowchart LR
    TRG["API / Schedule\nTrigger"]

    subgraph WF["Prefect Workflows (ConcurrentTaskRunner)"]
        PRF["profile_database\n(Profiling)"]
        ENR["enrich_metadata\n(Enrichment)"]
        DQG["generate_dq_rules\n(DQ Generation)"]
    end

    SEM["asyncio.Semaphore\n(task-level rate gate)"]
    LLM["LLM API\n(OpenAI)"]
    DB[("PostgreSQL\n(datareadiness)")]
    APR{"User\nApproval"}
    SC["Scorecard\nUpdate"]

    TRG --> PRF & ENR & DQG
    PRF & DQG --> SEM
    ENR --> SEM
    SEM --> LLM
    PRF & ENR & DQG --> DB
    DB --> APR
    APR -->|"approved"| SC

    classDef trigger fill:#56B4E9,stroke:#555555,color:#000
    classDef wf fill:#E69F00,stroke:#555555,color:#000
    classDef gate fill:#F0E442,stroke:#555555,color:#000
    classDef ext fill:#CC79A7,stroke:#555555,color:#000
    classDef store fill:#009E73,stroke:#555555,color:#000
    classDef decision fill:#F0E442,stroke:#555555,color:#000
    classDef terminal fill:#009E73,stroke:#555555,color:#000
    class TRG trigger
    class PRF,ENR,DQG wf
    class SEM gate
    class LLM ext
    class DB store
    class APR decision
    class SC terminal
```

### Data Profiling Flow

Analyzes connected data sources to produce statistical profiles:

| Setting         | Value                                                                                                                         |
| --------------- | ----------------------------------------------------------------------------------------------------------------------------- |
| **Flow name**   | `profile-database`                                                                                                            |
| **Concurrency** | `ConcurrentTaskRunner(max_workers=10)`                                                                                        |
| **Input**       | Data connection ID, schema/table targets (database-level targets are not supported, specify schema or table), sampling config |
| **Output**      | Column and table-level profiling results                                                                                      |

<Note>
  Profiling, Metadata Assessment, and DQ Assessment workflows require **schema-level or table-level targets**. Database-level targets (entire database) are not supported, specify individual schemas or tables to process.
</Note>

### Metadata Enrichment Flow

Uses LLMs to generate business descriptions and classifications:

| Setting         | Value                                                           |
| --------------- | --------------------------------------------------------------- |
| **Flow name**   | `enrich-metadata`                                               |
| **Concurrency** | `ConcurrentTaskRunner(max_workers=5)`                           |
| **Input**       | Data connection ID, profiling results                           |
| **Output**      | Business descriptions, PII classifications, DQ rule suggestions |

### DQ Rule Generation Flow

Generates data quality rules based on profiling data and LLM analysis:

| Setting         | Value                                                                   |
| --------------- | ----------------------------------------------------------------------- |
| **Flow name**   | `generate-dq-rules`                                                     |
| **Concurrency** | `ConcurrentTaskRunner` with `asyncio.Semaphore` for LLM calls           |
| **Input**       | Schema context, profiling results, enrichment metadata                  |
| **Output**      | Structured DQ rules per column                                          |
| **Config**      | `DQ_RULE_GEN_MAX_CONCURRENT_TABLES`, `DQ_RULE_GEN_MAX_CONCURRENT_RULES` |

DQ rule generation does **not** require a reference pack. Earlier versions passed an industry context / reference pack into the workflow; that input was removed because it caused crashes under Prefect 3.x. Generation runs purely from schema context, profiling results, and metadata.

## Workflow approval and the scorecard

When you approve ("Save") a profiling or enrichment workflow run, the platform applies two changes atomically:

1. **Tombstone affected metric snapshots.** Affected `metric_snapshots` rows are written with `numerator=0, denominator=0`. This invalidates the prior score so the scorecard does not display stale numbers while the next assessment is pending.
2. **Resolve all matching OPEN violations.** Every `OPEN` violation tied to an invalidated metric is closed in the same transaction, not just the most recent. Previously the resolve step closed only the latest violation, leaving older OPEN duplicates on the dashboard.

Net effect for users: scorecards update in real time. After Save, expect to see the affected scorecard cells flip to **N/A** ("pending next assessment") and any related violation entries disappear from the open-violations view.

### FQN scoping for violations

Violations are filtered by `asset_fqn` using an exact-or-child boundary: a filter for `db.schema.card` matches `db.schema.card` itself and any column-level child like `db.schema.card.amount`, but does **not** match sibling tables with prefix collisions like `db.schema.cardtransaction`. Earlier versions used a plain `startswith`, which produced false matches on sibling tables sharing a name prefix.

### Ingestion triggered by `asset.created`

When an ingestion run is triggered by an `asset.created` event (or kicked off manually with no schemas listed in `config`), the workflow ingests **all schemas** in the connection, scoped by `config.database`. Earlier versions only ingested the first schema returned by the connection, which led to silent data gaps when a connection exposed multiple schemas.

## Concurrency Model

Workflows use a **hybrid concurrency** model combining Prefect's task runner with asyncio semaphores:

### Flow-Level Concurrency

The `ConcurrentTaskRunner` controls how many table-level tasks run simultaneously:

```python theme={null}
@flow(task_runner=ConcurrentTaskRunner(max_workers=10))
async def profile_database(tables: list[str]):
    # Up to 10 tables profiled concurrently
    tasks = [profile_table.submit(table) for table in tables]
    results = await asyncio.gather(*[t.result() for t in tasks])
    return results
```

### Task-Level Concurrency

Within each task, `asyncio.Semaphore` controls fine-grained concurrency (e.g., LLM API calls):

```python theme={null}
@task
async def enrich_table(table_info):
    semaphore = asyncio.Semaphore(20)  # Max 20 concurrent LLM calls

    async def process_column(column):
        async with semaphore:
            return await llm_client.enrich(column)

    results = await asyncio.gather(
        *[process_column(col) for col in table_info.columns]
    )
    return results
```

| Level | Mechanism              | Visibility            |
| ----- | ---------------------- | --------------------- |
| Flow  | `ConcurrentTaskRunner` | Visible in Prefect UI |
| Task  | `asyncio.Semaphore`    | Internal to the task  |

<Tip>
  Configure concurrency limits via `config.yaml` rather than hardcoding. This allows tuning based on the source database capacity and LLM API rate limits.
</Tip>

## Workflow Management

### Prefect UI

The Prefect UI (port 4200) provides visibility into workflow execution:

* **Flow runs**: Status, duration, logs for each execution
* **Task runs**: Individual task status within a flow
* **Deployments**: Registered flow definitions and schedules
* **Work pool**: Worker status and queue depth

## Configuration

Workflow configuration is managed via **OmegaConf** with environment variable overrides:

```yaml theme={null}
# config.yaml
profiling:
  max_concurrent_tables: 10
  sample_threshold: 1000000  # Sample tables with more than 1M rows
  sample_size: 100000

enrichment:
  max_concurrent_tables: 5
  max_concurrent_llm_calls: 20
  model: "vertex_ai/gemini-3.5-flash"
  temperature: 0.1

dq_rules:
  max_concurrent_tables: 10
  max_concurrent_rules: 20
```

**Precedence:** Environment variables > `config.yaml` > defaults

Environment variables can be referenced in config via OmegaConf's `${oc.env:VAR_NAME, default}` syntax. The CRAFT default is Vertex AI — common env vars are `${oc.env:VERTEXAI_PROJECT}` / `${oc.env:VERTEXAI_LOCATION, global}` (also accepts the Google Cloud SDK standards `GOOGLE_CLOUD_PROJECT` / `GOOGLE_CLOUD_LOCATION`) with `GOOGLE_APPLICATION_CREDENTIALS` for ADC. Provider-specific keys (`OPENAI_API_KEY`, `ANTHROPIC_API_KEY`, `GEMINI_API_KEY`) are only required when the configured model routes to a non-Vertex provider.

## Error Handling

| Error Type                  | Handling                                                    |
| --------------------------- | ----------------------------------------------------------- |
| Database connection failure | Flow fails fast with descriptive error                      |
| LLM API rate limit          | Retry with exponential backoff via tenacity                 |
| LLM API timeout             | Retry up to 3 times, then skip column and continue          |
| Single table failure        | Log error, continue processing remaining tables             |
| Infrastructure failure      | Prefect marks flow as failed; manual retry available via UI |

## Kubernetes Deployment

In Kubernetes, Prefect workflows run as pods in the `kubernetes-pool` work pool:

| Component      | Deployment                                                                                                                                                                                                                                           |
| -------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| Prefect Server | StatefulSet with PostgreSQL backend                                                                                                                                                                                                                  |
| Prefect Worker | Deployment watching the kubernetes-pool                                                                                                                                                                                                              |
| Flow runs      | Ephemeral pods created per flow execution                                                                                                                                                                                                            |
| Secrets        | Injected via Helm `secretKeyRef`. For Vertex AI (CRAFT default): mount the ADC JSON via `GOOGLE_APPLICATION_CREDENTIALS` or use Workload Identity. For non-Vertex providers: `OPENAI_API_KEY` / `ANTHROPIC_API_KEY` / `GEMINI_API_KEY` per provider. |

## Next Steps

<CardGroup cols={2}>
  <Card title="Data Profiling" icon="magnifying-glass" href="/data-governance/data-profiling">
    Learn about the profiling workflow in detail.
  </Card>

  <Card title="Data Enrichment" icon="wand-magic-sparkles" href="/data-governance/data-enrichment">
    Understand the LLM-powered enrichment pipeline.
  </Card>

  <Card title="Data Source Setup" icon="database" href="/guides/data-source-setup">
    Connect a database to use with governance workflows.
  </Card>
</CardGroup>