# How to Use DVC Ingest to Convert Existing Pipelines to CMF

> Convert existing DVC pipelines to CMF using cmf dvc ingest. Migrate dvc.lock stages to MLMD executions with versioned artifacts for seamless tracking.

- Repository: [Hewlett Packard Enterprise/cmf](https://github.com/hewlettpackard/cmf)
- Tags: how-to-guide
- Published: 2026-03-03

---

**Use `cmf dvc ingest` to migrate existing DVC pipelines into CMF's ML-Metadata (MLMD) store, converting `dvc.lock` stages into tracked executions with versioned artifacts.**

The Hewlett Packard Enterprise Common Metadata Framework (CMF) provides a bridge command that imports existing DVC pipeline definitions without requiring you to rewrite your workflows. By using **DVC ingest**, you can convert existing DVC pipelines to CMF and immediately gain execution lineage tracking, artifact versioning, and optional Neo4j graph visualization while preserving your original DVC commands and dependencies.

## Understanding the CMF DVC Ingest Architecture

The ingestion process recreates your DVC pipeline topology inside CMF's metadata store by parsing `dvc.lock` and mapping each stage to CMF executions. The implementation spans several core modules in the `hewlettpackard/cmf` repository.

### CLI Entry Point and Argument Parsing

The command-line interface is defined in [`cmflib/cmf_commands_wrapper.py`](https://github.com/hewlettpackard/cmf/blob/main/cmflib/cmf_commands_wrapper.py). When you run `cmf dvc ingest -f <mlmd-file>`, the wrapper parses arguments and forwards them to the internal `_dvc_ingest` helper.

```python

# cmflib/cmf_commands_wrapper.py (lines 85-99)

def _dvc_ingest(self):
    parser = argparse.ArgumentParser(description='Ingest DVC pipeline')
    parser.add_argument('-f', '--file', required=False, default='./mlmd',
                       help='Path to MLMD file')
    args = parser.parse_args(self.args)
    dvc_ingest(args.file)

```

The `-f` flag defaults to `./mlmd`, specifying the SQLite MLMD store location.

### Core Ingestion Logic

The public `dvc_ingest` function in [`cmflib/cmf.py`](https://github.com/hewlettpackard/cmf/blob/main/cmflib/cmf.py) instantiates a `Cmf` object and initiates the ingestion workflow.

```python

# cmflib/cmf.py (lines 39-56)

def dvc_ingest(file_name: str):
    """Ingest DVC pipeline into CMF."""
    metawriter = Cmf(filename=file_name, pipeline_name="dvc_pipeline")
    # Internal processing continues...

```

This creates a new MLMD SQLite store if the file does not exist, then processes the `dvc.lock` file found in the current working directory.

### Reading Existing CMF Metadata

Before processing new data, the system checks for existing executions to avoid duplicates. The `CmfQuery` class queries the MLMD store to build a lookup table mapping DVC commands to execution IDs.

```python

# examples/active_learning/dvc_cmf_ingest.py (lines 66-73)

cmf_query = CmfQuery(args.cmf_filename)
for pipeline in cmf_query.get_pipeline_names():
    for stage in cmf_query.get_pipeline_stages(pipeline):
        executions = cmf_query.get_all_executions_in_stage(stage)
        # Build command-to-execution mapping

```

The script constructs a `cmd_exe` dictionary that maps the exact DVC command list (without the leading `python` interpreter) to a string formatted as `<execution_id>,<stage>,<pipeline>` (lines 85-90).

### Parsing the DVC Lock File

The ingestion engine loads `dvc.lock` and extracts stage definitions, commands, dependencies (`deps`), and outputs (`outs`).

```python

# examples/active_learning/dvc_cmf_ingest.py (lines 94-119)

with open("dvc.lock") as f:
    lock = yaml.load(f, Loader=yaml.FullLoader)

pipeline_dict = {}
for stage_name, stage_data in lock['stages'].items():
    cmd = stage_data['cmd'].split()
    # Normalize command and extract artifacts

```

This creates a `pipeline_dict` that mirrors the DVC pipeline structure while normalizing command strings for comparison.

### Matching or Creating Executions

For each stage in the pipeline, the system normalizes the command (removing the interpreter) and checks against existing executions:

- **If a match exists**: The corresponding execution is updated via `metawriter.update_execution` with newly captured DVC artifacts.
- **If no match exists**: A new execution is created using `metawriter.create_execution`, and lineage is recorded as `execution_name,context,pipeline` (lines 144-166).

```python

# Lineage handling example

cmd = stage_data['cmd'].split()[1:]  # Drop interpreter

lineage = cmd_exe.get(str(cmd))
if lineage:
    metawriter.ingest_metadata(lineage, stage_data, True, metawriter)
else:
    lineage = f"{cmd[0]},{stage_name},{metawriter.pipeline_name}"
    metawriter.ingest_metadata(lineage, stage_data, False, metawriter)

```

### Logging Artifacts with DVC Integration

For each dependency and output, `metawriter.log_dataset_with_version` integrates with DVC to capture versioning information. The system calls `commit_output` to ensure DVC tracks the artifact, then retrieves the DVC hash and URL via the DVC wrapper in [`cmflib/dvc_wrapper.py`](https://github.com/hewlettpackard/cmf/blob/main/cmflib/dvc_wrapper.py) (lines 99-106 and 111-118).

```python

# cmflib/dvc_wrapper.py integration

from cmflib.dvc_wrapper import dvc_get_hash, dvc_get_url

hash_value = dvc_get_hash(file_path)
url_value = dvc_get_url(file_path)

# Stored as URI in MLMD with custom properties (git_repo, Commit, url)

```

### Persisting the Lock File

After processing all stages, the original `dvc.lock` file itself is committed as an artifact using `metawriter.log_dvc_lock` (lines 667-678 in [`cmflib/cmf.py`](https://github.com/hewlettpackard/cmf/blob/main/cmflib/cmf.py)). This enables future re-ingestion without re-running DVC commands.

### Optional Neo4j Graph Visualization

When initialized with `graph=True`, the `Cmf` class mirrors each artifact, execution, and dataset as nodes and relationships in Neo4j via the `graph_wrapper` module, enabling visual pipeline exploration.

## Practical Implementation Examples

### One-Command CLI Ingestion

Run the following inside any directory containing a `dvc.lock` file:

```bash

# Use default ./mlmd output file

cmf dvc ingest

# Or specify a custom MLMD file name

cmf dvc ingest -f my_pipeline.mlmd

```

This executes the wrapper in [`cmflib/cmf_commands_wrapper.py`](https://github.com/hewlettpackard/cmf/blob/main/cmflib/cmf_commands_wrapper.py) and processes the pipeline immediately.

### Python API for Programmatic Control

For custom workflows, use the Python API directly:

```python
from cmflib.cmf import Cmf, dvc_ingest

# Method 1: High-level helper

dvc_ingest(file_name="my_pipeline.mlmd")

# Method 2: Manual control with Cmf object

metawriter = Cmf(
    filepath="my_pipeline.mlmd",
    pipeline_name="my_dvc_pipeline",
    graph=True
)
metawriter.log_dvc_lock("dvc.lock")

```

### Complete Ingestion Script

For production pipelines, implement the full ingestion pattern shown in the CMF examples:

```python
import argparse
import yaml
import uuid
from cmflib.cmf import Cmf
from cmflib.cmfquery import CmfQuery

parser = argparse.ArgumentParser()
parser.add_argument('--cmf_filename', type=str, default="mlmd")
args = parser.parse_args()

# Build existing execution lookup table

cmd_exe = {}
cmf_query = CmfQuery(args.cmf_filename)
for pipeline in cmf_query.get_pipeline_names():
    for stage in cmf_query.get_pipeline_stages(pipeline):
        for _, row in cmf_query.get_all_executions_in_stage(stage).iterrows():
            exe = row['Execution']
            if exe not in cmd_exe:
                cmd_exe[exe] = f"{row['id']},{stage},{pipeline}"

# Initialize CMF writer

metawriter = Cmf(
    filename="mlmd",
    pipeline_name=f"Pipeline-{uuid.uuid4()}",
    graph=True
)

# Process dvc.lock

with open("dvc.lock") as f:
    lock = yaml.load(f, Loader=yaml.FullLoader)

for stage_name, stage_data in lock['stages'].items():
    cmd = stage_data['cmd'].split()[1:]  # Remove interpreter

    lineage = cmd_exe.get(str(cmd))
    
    if lineage:
        # Update existing execution

        metawriter.ingest_metadata(lineage, stage_data, True, metawriter)
    else:
        # Create new execution

        lineage = f"{cmd[0]},{stage_name},{metawriter.pipeline_name}"
        metawriter.ingest_metadata(lineage, stage_data, False, metawriter)

# Commit lock file as artifact

metawriter.log_dvc_lock("dvc.lock")

```

This script is adapted from [`examples/active_learning/dvc_cmf_ingest.py`](https://github.com/hewlettpackard/cmf/blob/main/examples/active_learning/dvc_cmf_ingest.py) and demonstrates production-grade execution matching and metadata preservation.

## Summary

- **DVC ingest** converts existing DVC pipelines to CMF by parsing `dvc.lock` and mapping stages to MLMD executions.
- The command entry point lives in [`cmflib/cmf_commands_wrapper.py`](https://github.com/hewlettpackard/cmf/blob/main/cmflib/cmf_commands_wrapper.py), while core logic resides in [`cmflib/cmf.py`](https://github.com/hewlettpackard/cmf/blob/main/cmflib/cmf.py).
- The system prevents duplicate executions by matching normalized DVC commands against existing CMF metadata using `CmfQuery`.
- Artifacts are versioned using DVC hashes retrieved via [`cmflib/dvc_wrapper.py`](https://github.com/hewlettpackard/cmf/blob/main/cmflib/dvc_wrapper.py) and stored with Git commit metadata.
- The original `dvc.lock` file is preserved as an MLMD artifact via `log_dvc_lock` for reproducibility.
- Optional Neo4j graph support enables visual lineage tracking when `graph=True` is specified.

## Frequently Asked Questions

### What is the difference between `cmf dvc ingest` and running a native CMF pipeline?

**`cmf dvc ingest` imports existing DVC pipelines without code modification**, whereas native CMF pipelines require using the `Cmf` class decorators or context managers during execution. Ingest is ideal for teams migrating existing DVC workflows who want to preserve their current [`dvc.yaml`](https://github.com/hewlettpackard/cmf/blob/main/dvc.yaml) definitions while gaining CMF's metadata tracking capabilities.

### Does DVC ingest modify my original DVC files or pipeline structure?

**No, the ingestion process is read-only** regarding your DVC configuration. The command only reads `dvc.lock` to extract stage definitions, dependencies, and outputs. It writes metadata to the specified MLMD file (default `./mlmd`) without altering [`dvc.yaml`](https://github.com/hewlettpackard/cmf/blob/main/dvc.yaml), `dvc.lock`, or your data files.

### How does CMF handle DVC artifacts that already exist in the metadata store?

**CMF uses command-line signature matching to detect existing executions**. It normalizes DVC commands (removing the Python interpreter) and compares them against stored executions. If a match is found, `update_execution` merges new artifact information; otherwise, `create_execution` generates a new lineage entry. This prevents duplicate entries while allowing iterative pipeline development.

### Can I ingest DVC pipelines that use multiple stages and complex dependencies?

**Yes, the ingestion engine fully supports multi-stage pipelines**. As implemented in [`examples/active_learning/dvc_cmf_ingest.py`](https://github.com/hewlettpackard/cmf/blob/main/examples/active_learning/dvc_cmf_ingest.py), the system iterates through all stages defined in `dvc.lock`, processes each dependency and output independently, and maintains the topological relationships between stages in the MLMD store. Complex DAGs are preserved as execution lineages with proper artifact dependencies.