Data Productivity Cloud API - Pipeline Execution and Status Retrieval Framework

In the first three parts of this blog series, we walked through an overview of how to set up the Matillion Data Productivity Cloud (DPC) API using a Custom Connector to utilize those endpoints in your Designer Projects. In this final part, we will explore how to leverage that knowledge to build a framework within Designer to:

Execute pipelines using the Data Productivity Cloud APIs and store the pipelineExecutionIds into a persistent log table for each pipeline execution.
Retrieve the metadata for pipeline executions and update the log table with details of the execution metadata (Ex. startedAt, status, FinshedAt, message)

Prerequisites

Set up OAuth authentication for Data Productivity Cloud API in Manage Custom Connector.
Configure the below endpoints for the Data Productivity Cloud API in a Custom Connector
1. Execute Published Pipeline
2. Pipeline Status
Create an OAuth entry in the Designer project to use the above Data Productivity Cloud API Custom Connector endpoints.
Create a persistent Log table in Snowflake to store Pipeline Execution metadata information. Below is the DDL for the structure that was used for this blog.

	CREATE OR REPLACE TABLE
<DBName>.<schemaName>.PIPELINE_EXECUTION_LOG
(
"pipelineExecutionId" VARCHAR(255),
"projectId" VARCHAR(255),
"pipelineName" VARCHAR(255),
"environmentName" VARCHAR(255),
"startedAt" TIMESTAMP_TZ(9),
"status" VARCHAR(20),
"finshedAt" TIMESTAMP_TZ(9),
"message" VARCHAR(9999)
);

Building the Pipeline Execution Framework

Project Variable Setup

Add the following project variables to store the database, schema, and table name of the persistent log table.

Job Name: 1 - Execute Pipeline

The first process uses the Execute Published Pipeline endpoint to execute a published pipeline and append the metadata returned by the API endpoint (Ex. pipelineExecutionId) along with other relevant information used for the API (Ex. projectId, environmentName) to the log table.

Steps:

1.Generate Stage Table Name: Generate a stage table name during runtime that will be used to store the response from the Execute Published Pipeline endpoint. This table is temporarily created for use during this process and gets dropped in the last step of the process.

2. DPC API - Execute Published Pipeline: Executes a published pipeline via the DPC API and creates a stage Snowflake table to store the JSON response. The pipelineName, environment Name, and values for Pipeline Variables can be passed to this endpoint via the POST Body.

3. Run 1a - Append Pipeline Execution Id to Log Table: Runs a transformation to append the pipelineExecutionId of the published pipeline to the PIPELINE_EXECUTION_LOG table.

4. Drop Stage Table: This component runs a SQL script to drop the stage table created earlier in the pipeline after the data load completes.

Job Name: 2 - Refresh Pipeline Execution Status

The second process retrieves the pipeline execution status for each pipeline record in the Log table that is in RUNNING status (not yet finished) and captures the metadata to update the pipeline’s execution details.

The way this is designed, this process can be run or scheduled separately from the 1 - Execute Pipeline process.

Steps:

1.Table Iterator: Selects rows from the PIPELINE_EXECUTION_LOG table with status of RUNNING to extract column values and pass them as variables to the next component. This step also generates the stage table names that are used during the process to store the responses from the Pipeline Execution Status endpoint for each pipelineExecutionId.

2. Run 2a - Get Pipeline Execution Status: Runs an orchestration job to retrieve pipeline execution status, passing the extracted variable values. This runs iteratively for each pipelineExecutionId

3. DPC API - Get Pipeline Execution Status: This component extracts data from the Pipeline Status endpoint to retrieve the status of a pipeline execution, and writes the result to a Snowflake staging table.

4. Run 2b - Upsert Pipeline Execution Status to Log Table: Runs a transformation job that upserts the pipeline execution status retrieved from the API call into the PIPELINE_EXECUTION_LOG table.

5. Drop Stage Table: SQL script component that drops the stage table created in the previous component using the stage table name referenced by the variable.

Final thoughts

By following this article and leveraging the downloadable pipelines from the Matillion Exchange, you should now have a baseline architecture for a framework to capture pipeline execution history in Snowflake. We understand that each customer's implementation of this might be different, hence the intent of this framework is to provide a starting point to enhance or add on to the functionality, and customize it to specific requirements.

Catch up:

Part Two: Creating Custom Connector Endpoints for DPC API

Part Three: Setting up OAuth in Designer Project to use the DPC API Custom Connector Endpoints

talk to sales