Data Version Pipeline¶

Elevate Data empowers organizations to effectively manage their data lifecycle by enabling seamless integration of diverse data sources on a large scale, while also providing tools for enhanced governance and efficiency.

Elevate Data manages connections to various data sources (S3, Snowflake, and Databricks at this stage), allowing for efficient data extraction.

It also provides a central repository of data elements, imported from different sources, for greater transparency and efficient collaboration.

This notebook outlines how to import data from an external data source to Price-It on a fixed cadence:

1. Create a new version of an existing data table to import updated data from the data source.
2. Create a new version of an existing dataset that includes this data table and a recipe chain.
3. Update the version of the data table in the dataset to the latest.
4. Run the recipes on full data.
5. Release the dataset version.
6. Transfer dataset to predefined Price-It target.

Install the SDK¶

In [ ]:

skip-execution

!pip install --upgrade earnix-elevate

!pip install --upgrade earnix-elevate

Set your Elevate server and credentials¶

In this example we use environment variables to authenticate to Elevate, but you can also inject your credentials using Python arguments to each Service's client, like in the commented example.

In [ ]:

import os

os.environ.setdefault("E2_SERVER", "YOUR_SERVER")
os.environ.setdefault("E2_CLIENT_ID", "YOUR_CLIENT_ID")
os.environ.setdefault("E2_SECRET_KEY", "YOUR_SECRET_KEY")
# Alternatively:
# DataTableService(
#     server="YOUR_SERVER",
#     client_id="YOUR_CLIENT_ID",
#     secret_key="YOUR_SECRET_KEY"
# )

import os os.environ.setdefault("E2_SERVER", "YOUR_SERVER") os.environ.setdefault("E2_CLIENT_ID", "YOUR_CLIENT_ID") os.environ.setdefault("E2_SECRET_KEY", "YOUR_SECRET_KEY") # Alternatively: # DataTableService( # server="YOUR_SERVER", # client_id="YOUR_CLIENT_ID", # secret_key="YOUR_SECRET_KEY" # )

Imports and demo preparation¶

In [ ]:

from earnix_elevate import (
    CreateDatasetVersionRequest,
    CreateExportRequest,
    DatasetService,
    DataTableService,
    ExecuteRecipeRequest,
    ExportService,
    UpdateDatasetDataTableNodeRequest,
)

from earnix_elevate import ( CreateDatasetVersionRequest, CreateExportRequest, DatasetService, DataTableService, ExecuteRecipeRequest, ExportService, UpdateDatasetDataTableNodeRequest, )

Create Additional Data Version¶

You can create a new version of your data table to retrieve updated data from the source. This is useful when the underlying data has changed and you want to refresh your data table.

Create a Data Table service client¶

In [ ]:

data_table_service = DataTableService()

data_table_service = DataTableService()

Get the Data Table¶

In [ ]:

import os

DATA_TABLE_ID = int(
    os.environ.get("PIPELINE_DATA_TABLE_ID", "29104")
)  # use here your data table system id

data_table = data_table_service.get_data_table(DATA_TABLE_ID)
print(f"\n{data_table=}")

import os DATA_TABLE_ID = int( os.environ.get("PIPELINE_DATA_TABLE_ID", "29104") ) # use here your data table system id data_table = data_table_service.get_data_table(DATA_TABLE_ID) print(f"\n{data_table=}")

Create the Data Table Version¶

In [ ]:

# blocking=True creates the version and waits for its import to finish,
# returning the imported version (raising on failure/timeout).
data_table_version = data_table_service.create_data_version(
    data_table.id, blocking=True
)
print(f"\n{data_table_version=}")
data_table_id = data_table_version.data_table_id
data_table_version_number = data_table_version.version_number

# blocking=True creates the version and waits for its import to finish, # returning the imported version (raising on failure/timeout). data_table_version = data_table_service.create_data_version( data_table.id, blocking=True ) print(f"\n{data_table_version=}") data_table_id = data_table_version.data_table_id data_table_version_number = data_table_version.version_number

Create Additional Dataset Version¶

Each dataset version is represented by a transformation graph made up of nodes. This graph contains data nodes (data tables), recipe nodes (SQL transformations), and shows the relationships between them.

Create a Dataset service client¶

In [ ]:

dataset_service = DatasetService()

dataset_service = DatasetService()

Get the Dataset¶

In [ ]:

import os

DATASET_ID = int(
    os.environ.get("PIPELINE_DATASET_ID", "24307")
)  # use here your dataset system id

dataset = dataset_service.get_dataset(DATASET_ID)
print(f"\n{dataset=}")

dataset_id = dataset.id

import os DATASET_ID = int( os.environ.get("PIPELINE_DATASET_ID", "24307") ) # use here your dataset system id dataset = dataset_service.get_dataset(DATASET_ID) print(f"\n{dataset=}") dataset_id = dataset.id

Create the Dataset Version¶

In [ ]:

create_new_version_req = CreateDatasetVersionRequest(
    datasetId=dataset_id,
    sourceTransformationGraphVersionNumber=dataset.latest_transformation_graph.version_number,
)

dataset_version = dataset_service.create_new_version_to_dataset(
    dataset_id=dataset_id, create_dataset_version_request=create_new_version_req
)

print(f"\n{dataset_version=}")

transformation_graph = dataset_version.latest_transformation_graph
print(f"\n{transformation_graph=}")

create_new_version_req = CreateDatasetVersionRequest( datasetId=dataset_id, sourceTransformationGraphVersionNumber=dataset.latest_transformation_graph.version_number, ) dataset_version = dataset_service.create_new_version_to_dataset( dataset_id=dataset_id, create_dataset_version_request=create_new_version_req ) print(f"\n{dataset_version=}") transformation_graph = dataset_version.latest_transformation_graph print(f"\n{transformation_graph=}")

Update a Data Table version in a Dataset to the latest¶

In [ ]:

def get_data_table_node_from_transformation_graph():
    for node in transformation_graph.nodes:
        if node.node_type == "TABLE":
            return node
    return None


dataset_data_table_node = get_data_table_node_from_transformation_graph()
assert dataset_data_table_node

update_dataset_data_table_node_req = UpdateDatasetDataTableNodeRequest(
    data_table_id=data_table_id,
    data_table_version_number=data_table_version_number,
    dataset_id=dataset_id,
    id=dataset_data_table_node.id,
    lock_version=dataset_data_table_node.lock_version,
    node_type="TABLE",
)

dataset_data_table_node_with_updated_version = (
    dataset_service.update_dataset_node_in_transformation_graph(
        dataset_id=dataset_id,
        transformation_graph_version_number=transformation_graph.version_number,
        dataset_node_id=dataset_data_table_node.id,
        update_dataset_node_request=update_dataset_data_table_node_req,
    )
)

print(f"\n{dataset_data_table_node_with_updated_version=}")

def get_data_table_node_from_transformation_graph(): for node in transformation_graph.nodes: if node.node_type == "TABLE": return node return None dataset_data_table_node = get_data_table_node_from_transformation_graph() assert dataset_data_table_node update_dataset_data_table_node_req = UpdateDatasetDataTableNodeRequest( data_table_id=data_table_id, data_table_version_number=data_table_version_number, dataset_id=dataset_id, id=dataset_data_table_node.id, lock_version=dataset_data_table_node.lock_version, node_type="TABLE", ) dataset_data_table_node_with_updated_version = ( dataset_service.update_dataset_node_in_transformation_graph( dataset_id=dataset_id, transformation_graph_version_number=transformation_graph.version_number, dataset_node_id=dataset_data_table_node.id, update_dataset_node_request=update_dataset_data_table_node_req, ) ) print(f"\n{dataset_data_table_node_with_updated_version=}")

Run the Recipes on full data¶

• A Recipe is executed only if none of its predecessors in the chain have experienced failure during execution.

• When multiple Recipes are marked as pending, initiating the last Recipe will trigger the execution of all preceding Recipes. However, inspection data will be available solely for the last Recipe, and earlier Recipes will remain in a pending state.

Run the last Recipe¶

In [ ]:

run_req = ExecuteRecipeRequest(
    useSample=False,
)

# blocking=True runs the recipe and waits for it to finish, returning the
# terminal RecipeStatusResponse (raising on failure/timeout).
run_recipe = dataset_service.execute_recipe(
    dataset_id=dataset_id,
    transformation_graph_version_number=transformation_graph.version_number,
    dataset_node_id=transformation_graph.output_dataset_recipe_node_id,
    execute_recipe_request=run_req,
    blocking=True,
)

print(f"\nRecipe run completed\n{run_recipe=}")

run_req = ExecuteRecipeRequest( useSample=False, ) # blocking=True runs the recipe and waits for it to finish, returning the # terminal RecipeStatusResponse (raising on failure/timeout). run_recipe = dataset_service.execute_recipe( dataset_id=dataset_id, transformation_graph_version_number=transformation_graph.version_number, dataset_node_id=transformation_graph.output_dataset_recipe_node_id, execute_recipe_request=run_req, blocking=True, ) print(f"\nRecipe run completed\n{run_recipe=}")

Release the Dataset version¶

A Dataset can have two statuses:

• Draft: under construction (default)

• Released: finished and ready for export to Price-It

NOTE To release a Dataset, ensure that no Recipes have failed and that the Recipe processed the entire data. Only released Datasets can be exported to a Price-It project.

In [ ]:

released_dataset = dataset_service.release_transformation_graph(
    dataset_id=dataset_id,
    transformation_graph_version_number=transformation_graph.version_number,
)

print(f"\n{released_dataset=}")

released_dataset = dataset_service.release_transformation_graph( dataset_id=dataset_id, transformation_graph_version_number=transformation_graph.version_number, ) print(f"\n{released_dataset=}")

Transfer Dataset to predefined Price-It target¶

The export request supports two source types: 'datatable' and 'dataset', representing data tables and datasets, respectively. The source ID should correspond to the identifier of the data table or dataset to be exported.

Create an Export service client¶

In [ ]:

export_service = ExportService()

export_service = ExportService()

Export the Dataset¶

In [ ]:

import os

TARGET_ID = int(
    os.environ.get("PIPELINE_TARGET_ID", "9102")
)  # use here your target system id

export_dataset_request = CreateExportRequest(
    targetId=TARGET_ID,
    sourceType="dataset",
    sourceId=dataset_id,
    sourceVersion=transformation_graph.version_number,
    useSample=False,
    validateToRelease=True,
    castBooleanToInteger=False,
)

# blocking=True starts the export and waits for it to finish, returning the
# terminal ExportResponse (raising on failure/timeout).
export_response = export_service.export(export_dataset_request, blocking=True)
print(f"\nExport completed\n{export_response=}")

import os TARGET_ID = int( os.environ.get("PIPELINE_TARGET_ID", "9102") ) # use here your target system id export_dataset_request = CreateExportRequest( targetId=TARGET_ID, sourceType="dataset", sourceId=dataset_id, sourceVersion=transformation_graph.version_number, useSample=False, validateToRelease=True, castBooleanToInteger=False, ) # blocking=True starts the export and waits for it to finish, returning the # terminal ExportResponse (raising on failure/timeout). export_response = export_service.export(export_dataset_request, blocking=True) print(f"\nExport completed\n{export_response=}")