OSS Migration Playbook¶

# export_purview_glossary.py
import requests
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
token = credential.get_token("https://purview.azure.net/.default")

# Export glossary from Purview
purview_url = "https://{account}.purview.azure.com"
headers = {"Authorization": f"Bearer {token.token}"}

glossary = requests.get(
    f"{purview_url}/catalog/api/atlas/v2/glossary",
    headers=headers
).json()

# Transform for Atlas import
atlas_url = "http://atlas-service:21000"
atlas_auth = ("admin", "admin")

for term in glossary.get("terms", []):
    atlas_term = {
        "name": term["name"],
        "shortDescription": term.get("shortDescription", ""),
        "longDescription": term.get("longDescription", ""),
        "anchor": {"glossaryGuid": atlas_glossary_guid},
    }
    requests.post(
        f"{atlas_url}/api/atlas/v2/glossary/term",
        json=atlas_term,
        auth=atlas_auth
    )

# Create Atlas classification types
classification_defs = {
    "classificationDefs": [
        {
            "name": "PII_Name",
            "description": "Personally Identifiable Information - Name",
            "superTypes": ["PII"],
            "attributeDefs": []
        },
        {
            "name": "PII_Email",
            "description": "Personally Identifiable Information - Email",
            "superTypes": ["PII"],
            "attributeDefs": []
        },
        {
            "name": "PCI_CreditCard",
            "description": "PCI DSS - Credit Card Number",
            "superTypes": [],
            "attributeDefs": []
        }
    ]
}

requests.post(
    f"{atlas_url}/api/atlas/v2/types/typedefs",
    json=classification_defs,
    auth=atlas_auth
)

# Purview lineage → Atlas lineage
# Atlas uses "Process" entities to represent lineage connections

process_entity = {
    "entity": {
        "typeName": "Process",
        "attributes": {
            "name": "ETL_Pipeline_01",
            "qualifiedName": "etl_pipeline_01@csa_platform",
            "inputs": [{"typeName": "hdfs_path", "uniqueAttributes": {"qualifiedName": "adls://raw/data@csa"}}],
            "outputs": [{"typeName": "hdfs_path", "uniqueAttributes": {"qualifiedName": "adls://curated/data@csa"}}]
        }
    }
}

requests.post(
    f"{atlas_url}/api/atlas/v2/entity",
    json=process_entity,
    auth=atlas_auth
)

# superset_setup.py — Create database connections via Superset API
import requests

superset_url = "http://superset:8088"
session = requests.Session()

# Login
session.post(f"{superset_url}/api/v1/security/login", json={
    "username": "admin",
    "password": "admin",
    "provider": "db"
})

# Add Trino connection (replaces Power BI DirectQuery to Synapse)
session.post(f"{superset_url}/api/v1/database/", json={
    "database_name": "Trino - Data Lake",
    "engine": "trino",
    "sqlalchemy_uri": "trino://trino-coordinator:8080/delta",
    "extra": '{"engine_params": {"connect_args": {"http_scheme": "http"}}}',
    "expose_in_sqllab": True,
    "allow_ctas": False,
    "allow_cvas": True
})

# Add PostgreSQL connection (replaces Power BI Import mode)
session.post(f"{superset_url}/api/v1/database/", json={
    "database_name": "PostgreSQL - Operational",
    "engine": "postgresql",
    "sqlalchemy_uri": "postgresql+psycopg2://user:pass@pg-host:5432/csa_platform",
    "expose_in_sqllab": True
})

# migrate_search_index.py
import requests
from azure.identity import DefaultAzureCredential
from azure.search.documents.indexes import SearchIndexClient

# --- Source: Azure AI Search ---
credential = DefaultAzureCredential()
search_client = SearchIndexClient(
    endpoint="https://<search-service>.search.windows.us",
    credential=credential
)

# Get index definition
az_index = search_client.get_index("products")

# --- Target: OpenSearch ---
os_url = "https://opensearch:9200"
os_auth = ("admin", "admin")

# Map field types
type_map = {
    "Edm.String": "text",
    "Edm.Int32": "integer",
    "Edm.Int64": "long",
    "Edm.Double": "double",
    "Edm.Boolean": "boolean",
    "Edm.DateTimeOffset": "date",
    "Edm.GeographyPoint": "geo_point",
}

def map_fields(az_fields):
    properties = {}
    for field in az_fields:
        os_type = type_map.get(field.type, "text")
        prop = {"type": os_type}

        # Searchable strings get both text and keyword
        if field.type == "Edm.String" and field.searchable:
            prop = {
                "type": "text",
                "fields": {"keyword": {"type": "keyword", "ignore_above": 256}}
            }
        # Filterable-only strings become keyword
        elif field.type == "Edm.String" and field.filterable and not field.searchable:
            prop = {"type": "keyword"}

        # Vector fields
        if "Collection(Edm.Single)" in str(field.type):
            prop = {
                "type": "knn_vector",
                "dimension": field.vector_search_dimensions or 1536,
                "method": {
                    "name": "hnsw",
                    "space_type": "cosinesimil",
                    "engine": "nmslib"
                }
            }

        properties[field.name] = prop
    return properties

# Create OpenSearch index
os_index = {
    "settings": {
        "index": {
            "number_of_shards": 3,
            "number_of_replicas": 1,
            "knn": True
        }
    },
    "mappings": {
        "properties": map_fields(az_index.fields)
    }
}

requests.put(
    f"{os_url}/products",
    json=os_index,
    auth=os_auth,
    verify=False
)

# dags/etl_raw_to_curated.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.microsoft.azure.sensors.wasb import WasbBlobSensor
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

default_args = {
    "owner": "csa-platform",
    "depends_on_past": False,
    "email_on_failure": True,
    "email": ["platform-team@contoso.com"],
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
}

with DAG(
    dag_id="etl_raw_to_curated",
    default_args=default_args,
    description="Process raw data to curated zone (replaces ADF Copy + Data Flow)",
    schedule="0 6 * * *",      # Daily at 6 AM — was ADF Schedule Trigger
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["etl", "csa-platform"],
) as dag:

    # Sensor: wait for source file (replaces ADF Event Trigger)
    wait_for_data = WasbBlobSensor(
        task_id="wait_for_raw_data",
        container_name="raw",
        blob_name="data/incoming/{{ ds }}/",
        wasb_conn_id="azure_adls",
        timeout=3600,
        poke_interval=60,
    )

    # Transform: replaces ADF Data Flow
    def transform_data(**context):
        """Read raw CSV, apply transformations, write Parquet to curated."""
        import pandas as pd
        from azure.storage.filedatalake import DataLakeServiceClient

        ds = context["ds"]
        # ... transformation logic ...
        print(f"Processed data for {ds}")

    transform = PythonOperator(
        task_id="transform_raw_to_curated",
        python_callable=transform_data,
    )

    # Catalog: register in Atlas (replaces Purview auto-scan)
    def register_in_atlas(**context):
        """Register processed dataset in Apache Atlas catalog."""
        import requests
        atlas_url = "http://atlas-service:21000"
        # ... entity creation ...

    catalog = PythonOperator(
        task_id="register_in_catalog",
        python_callable=register_in_atlas,
    )

    wait_for_data >> transform >> catalog

# dags/orchestrator.py
from airflow import DAG
from airflow.operators.trigger_dagrun import TriggerDagRunOperator
from airflow.sensors.external_task import ExternalTaskSensor
from datetime import datetime

with DAG(
    dag_id="master_orchestrator",
    schedule="0 5 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["orchestration"],
) as dag:

    # Replaces ADF "Execute Pipeline" activity
    trigger_ingestion = TriggerDagRunOperator(
        task_id="trigger_ingestion",
        trigger_dag_id="etl_raw_to_curated",
        wait_for_completion=True,
    )

    trigger_quality = TriggerDagRunOperator(
        task_id="trigger_quality_checks",
        trigger_dag_id="data_quality_checks",
        wait_for_completion=True,
    )

    trigger_analytics = TriggerDagRunOperator(
        task_id="trigger_analytics",
        trigger_dag_id="analytics_refresh",
        wait_for_completion=True,
    )

    trigger_ingestion >> trigger_quality >> trigger_analytics