How to Use DVC Ingest to Convert Existing Pipelines to CMF

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. When you run cmf dvc ingest -f <mlmd-file>, the wrapper parses arguments and forwards them to the internal _dvc_ingest helper.

# 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 instantiates a Cmf object and initiates the ingestion workflow.

# 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.

# 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).

# 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).

# 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 (lines 99-106 and 111-118).

# 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). 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:

# 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 and processes the pipeline immediately.

Python API for Programmatic Control

For custom workflows, use the Python API directly:

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:

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 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, while core logic resides in 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 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 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, 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, 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}