Pipeline¶

{
  "$id": "https://open-metadata.org/schema/entity/data/pipeline.json",
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "Pipeline",
  "$comment": "@om-entity-type",
  "description": "This schema defines the Pipeline entity. A pipeline enables the flow of data from source to destination through a series of processing steps. ETL is a type of pipeline where the series of steps Extract, Transform and Load the data.",
  "type": "object",
  "javaType": "org.openmetadata.schema.entity.data.Pipeline",
  "javaInterfaces": ["org.openmetadata.schema.EntityInterface"],

  "definitions": {
    "statusType": {
      "javaType": "org.openmetadata.schema.type.StatusType",
      "description": "Enum defining the possible Status.",
      "type": "string",
      "enum": ["Successful", "Failed", "Pending", "Skipped"],
      "javaEnums": [
        {"name": "Successful"},
        {"name": "Failed"},
        {"name": "Pending"},
        {"name": "Skipped"}
      ]
    },
    "pipelineState": {
      "description": "Enum defining the possible Pipeline State.",
      "type": "string",
      "enum": ["Active", "Inactive"],
      "javaEnums": [
        {"name": "Active"},
        {"name": "Inactive"}
      ]
    },
    "taskStatus": {
      "type": "object",
      "javaType": "org.openmetadata.schema.type.Status",
      "description": "This schema defines a time series of the status of a Pipeline or Task.",
      "properties": {
        "name": {"description": "Name of the Task.", "type": "string"},
        "executionStatus": {"description": "Status at a specific execution date.", "$ref": "#/definitions/statusType"},
        "startTime": {"description": "Task start time", "$ref": "../../type/basic.json#/definitions/timestamp"},
        "endTime": {"description": "Task end time", "$ref": "../../type/basic.json#/definitions/timestamp"},
        "logLink": {"description": "Task end time", "type": "string", "format": "uri"},
        "taskId": {"description": "Task ID from orchestrator", "type": "string"},
        "transformationType": {"description": "Type of transformation", "type": "string", "enum": ["filter", "aggregate", "join", "pivot", "custom", "extract", "load"]},
        "transformationLogic": {"description": "SQL or code executed", "type": "string"},
        "inputs": {"description": "Task-specific inputs", "type": "array", "items": {"$ref": "#/definitions/datasetUsage"}},
        "outputs": {"description": "Task-specific outputs", "type": "array", "items": {"$ref": "#/definitions/datasetGeneration"}},
        "metrics": {"$ref": "#/definitions/taskMetrics"}
      },
      "required": ["name", "executionStatus"]
    },
    "task": {
      "type": "object",
      "javaType": "org.openmetadata.schema.type.Task",
      "properties": {
        "name": {"description": "Name that identifies this task instance uniquely.", "type": "string"},
        "displayName": {"description": "Display Name that identifies this Task.", "type": "string"},
        "fullyQualifiedName": {"description": "A unique name that identifies a pipeline in the format 'ServiceName.PipelineName.TaskName'.", "type": "string"},
        "description": {"description": "Description of this Task.", "$ref": "../../type/basic.json#/definitions/markdown"},
        "sourceUrl": {"description": "Task URL to visit/manage.", "$ref": "../../type/basic.json#/definitions/sourceUrl"},
        "downstreamTasks": {"description": "All the tasks that are downstream of this task.", "type": "array", "items": {"type": "string"}},
        "taskType": {"description": "Type of the Task.", "type": "string"},
        "taskSQL": {"description": "SQL used in the task.", "$ref": "../../type/basic.json#/definitions/sqlQuery"},
        "startDate": {"description": "start date for the task.", "type": "string"},
        "endDate": {"description": "end date for the task.", "type": "string"},
        "tags": {"description": "Tags for this task.", "type": "array", "items": {"$ref": "../../type/tagLabel.json"}},
        "owners": {"description": "Owners of this task.", "$ref": "../../type/entityReferenceList.json"}
      },
      "required": ["name"]
    },
    "pipelineStatus": {
      "description": "Series of pipeline executions, its status and task status.",
      "type": "object",
      "properties": {
        "timestamp": {"description": "Timestamp where the job was executed.", "$ref": "../../type/basic.json#/definitions/timestamp"},
        "executionStatus": {"description": "Status at a specific execution date.", "$ref": "#/definitions/statusType"},
        "taskStatus": {"description": "Series of task executions and its status.", "type": "array", "items": {"$ref": "#/definitions/taskStatus"}},
        "executionId": {"description": "External execution ID from the orchestrator", "type": "string"},
        "endTime": {"description": "Execution end time", "$ref": "../../type/basic.json#/definitions/timestamp"},
        "version": {"description": "Pipeline version that was executed", "type": "string"},
        "parameters": {"description": "Execution parameters/configuration", "type": "object"},
        "executedBy": {"description": "User or service that triggered the execution", "$ref": "../../type/entityReference.json"},
        "inputs": {"description": "Input datasets consumed by this execution", "type": "array", "items": {"$ref": "#/definitions/datasetUsage"}},
        "outputs": {"description": "Output datasets generated by this execution", "type": "array", "items": {"$ref": "#/definitions/datasetGeneration"}},
        "metrics": {"description": "Execution metrics", "$ref": "#/definitions/executionMetrics"},
        "error": {"description": "Error details if execution failed", "$ref": "#/definitions/executionError"}
      },
      "required": ["timestamp", "executionStatus"]
    }
  },

  "properties": {
    "id": {
      "description": "Unique identifier that identifies a pipeline instance.",
      "$ref": "../../type/basic.json#/definitions/uuid"
    },
    "name": {
      "description": "Name that identifies this pipeline instance uniquely.",
      "$ref": "../../type/basic.json#/definitions/entityName"
    },
    "displayName": {
      "description": "Display Name that identifies this Pipeline. It could be title or label from the source services.",
      "type": "string"
    },
    "fullyQualifiedName": {
      "description": "A unique name that identifies a pipeline in the format 'ServiceName.PipelineName'.",
      "$ref": "../../type/basic.json#/definitions/fullyQualifiedEntityName"
    },
    "description": {
      "description": "Description of this Pipeline.",
      "$ref": "../../type/basic.json#/definitions/markdown"
    },
    "dataProducts": {
      "description": "List of data products this entity is part of.",
      "$ref": "../../type/entityReferenceList.json"
    },
    "version": {
      "description": "Metadata version of the entity.",
      "$ref": "../../type/entityHistory.json#/definitions/entityVersion"
    },
    "updatedAt": {
      "description": "Last update time corresponding to the new version of the entity in Unix epoch time milliseconds.",
      "$ref": "../../type/basic.json#/definitions/timestamp"
    },
    "updatedBy": {
      "description": "User who made the update.",
      "type": "string"
    },
    "impersonatedBy": {
      "description": "Bot user that performed the action on behalf of the actual user.",
      "$ref": "../../type/basic.json#/definitions/impersonatedBy"
    },
    "sourceUrl": {
      "description": "Pipeline URL to visit/manage. This URL points to respective pipeline service UI.",
      "$ref": "../../type/basic.json#/definitions/sourceUrl"
    },
    "concurrency": {
      "description": "Concurrency of the Pipeline.",
      "type": "integer"
    },
    "pipelineLocation": {
      "description": "Pipeline Code Location.",
      "type": "string"
    },
    "startDate": {
      "description": "Start date of the workflow.",
      "$ref": "../../type/basic.json#/definitions/dateTime"
    },
    "endDate": {
      "description": "End date of the workflow.",
      "$ref": "../../type/basic.json#/definitions/dateTime"
    },
    "tasks": {
      "description": "All the tasks that are part of pipeline.",
      "type": "array",
      "items": {
        "$ref": "#/definitions/task"
      }
    },
    "pipelineStatus": {
      "description": "Latest Pipeline Status.",
      "$ref": "#/definitions/pipelineStatus"
    },
    "state": {
      "description": "State of the Pipeline.",
      "$ref": "#/definitions/pipelineState"
    },
    "followers": {
      "description": "Followers of this Pipeline.",
      "$ref": "../../type/entityReferenceList.json"
    },
    "tags": {
      "description": "Tags for this Pipeline.",
      "type": "array",
      "items": {
        "$ref": "../../type/tagLabel.json"
      }
    },
    "href": {
      "description": "Link to the resource corresponding to this entity.",
      "$ref": "../../type/basic.json#/definitions/href"
    },
    "owners": {
      "description": "Owners of this pipeline.",
      "$ref": "../../type/entityReferenceList.json"
    },
    "service": {
      "description": "Link to service where this pipeline is hosted in.",
      "$ref": "../../type/entityReference.json"
    },
    "serviceType": {
      "description": "Service type where this pipeline is hosted in.",
      "$ref": "../services/pipelineService.json#/definitions/pipelineServiceType"
    },
    "usageSummary": {
      "description": "Latest usage information for this pipeline.",
      "$ref": "../../type/usageDetails.json"
    },
    "changeDescription": {
      "description": "Change that lead to this version of the entity.",
      "$ref": "../../type/entityHistory.json#/definitions/changeDescription"
    },
    "incrementalChangeDescription": {
      "description": "Change that lead to this version of the entity.",
      "$ref": "../../type/entityHistory.json#/definitions/changeDescription"
    },
    "deleted": {
      "description": "When `true` indicates the entity has been soft deleted.",
      "type": "boolean",
      "default": false
    },
    "extension": {
      "description": "Entity extension data with custom attributes added to the entity.",
      "$ref": "../../type/basic.json#/definitions/entityExtension"
    },
    "scheduleInterval": {
      "description": "Scheduler Interval for the pipeline in cron format.",
      "type": "string"
    },
    "domains": {
      "description": "Domains the Pipeline belongs to. When not set, the pipeline inherits the domain from the Pipeline service it belongs to.",
      "$ref": "../../type/entityReferenceList.json"
    },
    "votes": {
      "description": "Votes on the entity.",
      "$ref": "../../type/votes.json"
    },
    "lifeCycle": {
      "description": "Life Cycle properties of the entity",
      "$ref": "../../type/lifeCycle.json"
    },
    "certification": {
      "$ref": "../../type/assetCertification.json"
    },
    "sourceHash": {
      "description": "Source hash of the entity",
      "type": "string",
      "minLength": 1,
      "maxLength": 32
    },
    "entityStatus": {
      "description": "Status of the Pipeline.",
      "$ref": "../../type/status.json"
    }
  },

  "required": ["id", "name", "service"],
  "additionalProperties": false
}

{
  "@context": {
    "@vocab": "https://open-metadata.org/schema/",
    "om": "https://open-metadata.org/schema/",
    "rdfs": "http://www.w3.org/2000/01/rdf-schema#",
    "xsd": "http://www.w3.org/2001/XMLSchema#",

    "Pipeline": "om:Pipeline",
    "name": {
      "@id": "om:pipelineName",
      "@type": "xsd:string"
    },
    "fullyQualifiedName": {
      "@id": "om:fullyQualifiedName",
      "@type": "xsd:string"
    },
    "displayName": {
      "@id": "om:displayName",
      "@type": "xsd:string"
    },
    "description": {
      "@id": "om:description",
      "@type": "xsd:string"
    },
    "pipelineUrl": {
      "@id": "om:pipelineUrl",
      "@type": "xsd:anyURI"
    },
    "scheduleInterval": {
      "@id": "om:scheduleInterval",
      "@type": "xsd:string"
    },
    "startDate": {
      "@id": "om:startDate",
      "@type": "xsd:dateTime"
    },
    "endDate": {
      "@id": "om:endDate",
      "@type": "xsd:dateTime"
    },
    "state": {
      "@id": "om:pipelineState",
      "@type": "@vocab"
    },
    "concurrency": {
      "@id": "om:concurrency",
      "@type": "xsd:integer"
    },
    "pipelineLocation": {
      "@id": "om:pipelineLocation",
      "@type": "xsd:string"
    },
    "pipelineStatus": {
      "@id": "om:pipelineStatus",
      "@type": "@id"
    },
    "tasks": {
      "@id": "om:hasTask",
      "@type": "@id",
      "@container": "@list"
    },
    "service": {
      "@id": "om:belongsToPipelineService",
      "@type": "@id"
    },
    "serviceType": {
      "@id": "om:serviceType",
      "@type": "xsd:string"
    },
    "owners": {
      "@id": "om:pipelineOwnedBy",
      "@type": "@id",
      "@container": "@set"
    },
    "domains": {
      "@id": "om:pipelineInDomain",
      "@type": "@id",
      "@container": "@set"
    },
    "followers": {
      "@id": "om:pipelineFollowedBy",
      "@type": "@id",
      "@container": "@set"
    },
    "tags": {
      "@id": "om:pipelineHasTag",
      "@type": "@id",
      "@container": "@set"
    },
    "dataProducts": {
      "@id": "om:partOfDataProduct",
      "@type": "@id",
      "@container": "@set"
    },
    "votes": {
      "@id": "om:hasVotes",
      "@type": "@id"
    },
    "lifeCycle": {
      "@id": "om:hasLifeCycle",
      "@type": "@id"
    },
    "certification": {
      "@id": "om:hasCertification",
      "@type": "@id"
    }
  }
}

{
  "@context": "https://open-metadata.org/context/pipeline.jsonld",
  "@type": "Pipeline",
  "@id": "https://example.com/pipelines/customer_etl",

  "name": "customer_etl",
  "fullyQualifiedName": "airflow_prod.customer_etl",
  "displayName": "Customer ETL Pipeline",
  "description": "Daily ETL pipeline for customer data",
  "sourceUrl": "https://github.com/company/pipelines/blob/main/dags/customer_etl.py",

  "scheduleInterval": "0 2 * * *",
  "startDate": "2024-01-01T00:00:00Z",
  "endDate": "2024-12-31T23:59:59Z",
  "state": "Active",
  "concurrency": 1,
  "pipelineLocation": "/opt/airflow/dags/customer_etl.py",

  "pipelineStatus": {
    "timestamp": 1704240000000,
    "executionStatus": "Successful",
    "executionId": "manual__2024-01-03T02:00:00+00:00"
  },

  "service": {
    "@id": "https://example.com/services/airflow_prod",
    "@type": "PipelineService",
    "name": "airflow_prod"
  },

  "serviceType": "Airflow",

  "owners": [
    {
      "@id": "https://example.com/teams/data-engineering",
      "@type": "Team",
      "name": "data-engineering",
      "displayName": "Data Engineering"
    }
  ],

  "domains": [
    {
      "@id": "https://example.com/domains/CustomerData",
      "@type": "Domain",
      "name": "CustomerData"
    }
  ],

  "followers": [
    {
      "@id": "https://example.com/users/john.doe",
      "@type": "User",
      "name": "john.doe"
    }
  ],

  "dataProducts": [
    {
      "@id": "https://example.com/dataProducts/CustomerInsights",
      "@type": "DataProduct",
      "name": "CustomerInsights"
    }
  ],

  "tags": [
    {
      "@id": "https://open-metadata.org/tags/Tier/Gold",
      "tagFQN": "Tier.Gold"
    },
    {
      "@id": "https://open-metadata.org/tags/Schedule/Daily",
      "tagFQN": "Schedule.Daily"
    }
  ],

  "tasks": [
    {
      "@type": "Task",
      "@id": "https://example.com/pipelines/customer_etl/tasks/extract_customers",
      "name": "extract_customers",
      "taskType": "PythonOperator"
    },
    {
      "@type": "Task",
      "@id": "https://example.com/pipelines/customer_etl/tasks/transform_customers",
      "name": "transform_customers",
      "taskType": "SparkSubmitOperator"
    },
    {
      "@type": "Task",
      "@id": "https://example.com/pipelines/customer_etl/tasks/load_customers",
      "name": "load_customers",
      "taskType": "PostgresOperator"
    }
  ]
}