Operator Integration

DAVE allows the execution of python scripts in the form of blocks, called operators. These blocks can be added into the environment as long as these two requirements are fullfilled:

  • A javascript file with the operator’s metadata

  • A python script to be executed as an airflow operator

This tutorial shows how to create custom operators, and how to add them to the DAE. The first section contains information regarding the operator metadata, such as the structure and parameters, and the second section is about the script itself.

Operator Metadata

This section contains the main body of an operator’s metadata. Below there is a code snippet with what parameters/values are required to write this metadata. The table contains a description of the ALL_CAPS_VALUES, including examples. For a complete structure of a operator’s metadata file please refer to this template. All of the operator metadata files go into /mongodb/docker-entrypoint-initdb.d.

db.operator_meta.insertOne(
    {
        op_name: OPERATOR_NAME,
        op_pres_name: OPERATOR_PRESENTATION_NAME,
        op_pres_desc: OPERATOR_DESCRIPTION,
        op_type: OPERATOR_TYPE,
        airflow_op_type: AIRFLOW_OPERATOR_TYPE,
        visibility: OPERATOR_VISIBILITY,
        poweredby: OPERATOR_TECHNOLOGY,
        info: OPERATOR_DETAILS,
        op_params: OPERATOR_INPUT_PARAMETERS
    }
)
Field Type Description Accepted Values Example

OPERATOR_NAME

string

Internal identifier for the operator

Any string, using _ for spaces

spark_file_connector

OPERATOR_PRESENTATION_NAME

string

Display name shown in the UI

Any string

File Connector

OPERATOR_DESCRIPTION

string

Short description of what the operator does

Any string

Connector for single files

AIRFLOW_OPERATOR_TYPE

string

Airflow operator class used to execute the task. Supports PythonOperator and PythonSensor

  • airflow.operators.python.PythonOperator

  • airflow.sensors.python.PythonSensor

airflow.operators.python.PythonOperator

OPERATOR_VISIBILITY

string

Whether the operator is public or private (to be implemented)

  • public

  • private

public

OPERATOR_TECHNOLOGY

string

Underlying technology powering the operator

Single word string

spark

OPERATOR_DETAILS

Object

Documentation for this operator (to be implemented)

See [operator-details]

(to be implemented)

OPERATOR_INPUT_PARAMETERS

Object

Input parameters required by the operator

See [operator-parameters]

See [operator-parameters]

OPERATOR_DETAILS

This section shows the content of the "info" value, referenced in the previous section. The contents for this part of the metadata is not mandatory, as they are mainly descriptions of both the operator, and its parameters. Below is the code snippet of this metadata section, as well as the table with a detailed description of each parameter.

{
    desc: OPERATOR_DETAILED_DESCRIPTION,
    parameters: OPERATOR_PARAMETERS_DESCRIPTION,
    advanced_parameters: OPERATOR_ADVANCED_PARAMETERS
}
Field Type Description Accepted Values Example

OPERATOR_DETAILED_DESCRIPTION

string

Detailed description of the operator’s functionality. Can be empty as it is not yet implemented.

—

""

OPERATOR_PARAMETERS_DESCRIPTION

Object

Descriptions of the operator’s basic inputs. Can be empty as it is not yet implemented.

—

{}

OPERATOR_ADVANCED_PARAMETERS

Object

Descriptions of the operator’s advanced inputs. Can be empty as it is not yet implemented.

—

{}

OPERATOR_PARAMETERS

The OPERATOR_PARAMETERS corresponds to the main parameters for a Apache Airflow python operator. This type of operator only requires 2 mandatory parameters, them being the "python_callable_file" and "python_callable_file", while the "op_kwargs" parameter is used to receive user inputs. Below is the code snippet for this section of the operator metadata, as well as a description of each parameter in the table.

{
    python_callable_file: OPERATOR_FILE_PATH,
    python_callable_name: OPERATOR_FUNCTION_NAME,
    op_kwargs: OPERATOR_INPUT_PARAMETERS
}
Field Type Description Accepted Values Example

OPERATOR_FILE_PATH

string

Path to the Python file to be executed

Path string

/opt/airflow/plugins/operators/connectors/file_v2.py

OPERATOR_FUNCTION_NAME

string

Name of the function used as the script entry point

Function name string

main

OPERATOR_INPUT_PARAMETERS

Array[Object]

List of objects describing each user input parameter

See [operator-input-parameters]

See [operator-input-parameters]

OPERATOR_INPUT_PARAMETERS

This section shows what are all the available parameters to define an user input. It is not mandatory to have all the shown parameters, since it depends of the type of input. Below is the code snippet with the template with all the parameters, as well as the table with a detailed description of each parameter.

op_kwargs: [
    {
        name: OPERATOR_INPUT_PARAMETER_NAME,
        pres_name: OPERATOR_INPUT_PARAMETER_PRESENTATION_NAME,
        pres_desc: OPERATOR_INPUT_PARAMETER_DESCRIPTION,
        type: OPERATOR_INPUT_PARAMETER_TYPE,
        params: OPERATOR_INPUT_PARAMETER_EXTERNAL_CONFIGURATION_STORAGE_PARAMS,
        accepted_values: OPERATOR_INPUT_PARAMETER_ACCEPTED_VALUES,
        default: OPERATOR_INPUT_PARAMETER_DEFAULT
    }
]
Field Type Description Accepted Values Example

name

string

Unique internal name for the input variable

String without special characters (except _)

"host"

pres_name

string

Display name shown in the UI

Any string

"Host"

pres_desc

string

Description of what the input variable represents

Any string

"Host of the relational database"

type

string

Data type of the input variable

  • string

  • integer

  • float

  • enum_string

  • enum_string_external

"string"

accepted_values

Array[string]

List of values available when type is enum_string or enum_string_external

Array of strings

["read", "write"]

default (optional)

any

Default value for the input

—

"read"

params

Object

Additional key-value configuration, used when the operator is an external service

Key-value pairs

See mongodb/docker-entrypoint-initdb.d/harmonization_external_service.js

Operator Script

The python script that contains the methods to perform the operator task needs to follow some rules in order to be integrated as an operator, not only to be used as an operator but also to be able to "interact" with the other operators.

The templates containing these rules are under /docs/templates, and are divided as follows:

All of the operator scripts go into either the analytics folder, the connectors folder or the transformers folder, depending of the type of operator. These folder are under the /airflow/plugins/operators/ path.

SDK Functions

DAVE provides an SDK with three helper modules to assist developers in the creation of operators:

1. mlflow_sdk.py

This module integrates with MLflow to facilitate experiment tracking, model logging, and parameter management.

Functions:

save_model_mlflow(model, model_name, flavour, params=None, metrics=None): Saves a machine learning model in MLflow.

load_model_mlflow(model_name): Loads a registered MLflow model.

Example Usage:

from dave_sdk.mlflow_sdk import save_model_mlflow, load_model_mlflow

# Save a model
save_model_mlflow(model, "my_model", "sklearn")

# Load a model
model = load_model_mlflow("my_model")

2. pandas_sdk.py

This module provides utility functions for working with Pandas dataframes, including data transformation and MongoDB storage.

The MongoDB collection and database where the dataframes will be saved in or loaded from is passed to the functions though the kwargs variable.

The dataframe is saved to the user’s MongoDB database in a collection with the following name structure: [workflow_name].[operator_name].[workflow_execution_date].

Functions:

save_pandas_df(df, kwargs): Saves a Pandas DataFrame to a MongoDB collection in the user’s Database.

load_pandas_df(kwargs): Loads a Pandas DataFrame from a MongoDB collection in the user’s Database.

Example Usage:

from dave_sdk.pandas_sdk import save_pandas_df, load_pandas_df

# Save DataFrame to MongoDB
save_pandas_df(df, kwargs)

# Load DataFrame from MongoDB
df = load_pandas_df(kwargs)

3. spark_sdk.py

This module integrates with Apache Spark to facilitate distributed data processing and MongoDB storage.

The MongoDB collection and database where the dataframes will be saved in or loaded from is passed to the functions though the kwargs variable.

The dataframe is saved to the user’s MongoDB database in a collection with the following name structure: [workflow_name].[operator_name].[workflow_execution_date].

Functions:

drop_null_columns(df): Drops columns with all null values from a Spark DataFrame.

save_spark_df(df, kwargs): Saves a Spark DataFrame to a MongoDB collection in the user’s Database.

load_spark_df(spark, kwargs): Loads a Spark DataFrame from a MongoDB collection in the user’s Database.

Example Usage:

from dave_sdk.spark_sdk import save_spark_df, load_spark_df, drop_null_columns

# Save Spark DataFrame to MongoDB
save_spark_df(df, kwargs)

# Load Spark DataFrame from MongoDB
df = load_spark_df(spark, kwargs)

# Drop null columns
df = drop_null_columns(df)

Create basic operator tutorial

This section contains an example of a simple operator to guide developers to create new operators. The following operator is extremely simple, it takes an input string from the user and prints it.

The python file is the following:

def main(**kwargs):
    stringtoprint = kwargs["string_to_print"]
    print(stringtoprint)

if __name__ == "__main__":
    main()

In this code the string is received by the kwargs.

The javascript file with the operator’s metadata is the following:

//This following line is standard to all operators that points to the db where the metadata is stored
DB_NAME = process.env.MONGO_INITDB_DATABASE
db = db.getSiblingDB(DB_NAME)

db.operator_meta.insertOne(
    {
        op_name:"print_string",
        op_pres_name:"Print String",
        op_pres_desc:"Example operator that prints a string provided by the user",
        op_type:"transformer",
        airflow_op_type: "airflow.operators.python.PythonOperator",
        visibility:"public",
        poweredby: "python",
        info: {
            desc: "",
            parameters:{
            },
            advanced_parameters:{

            }
        },
        op_params:{
            python_callable_file:"/opt/airflow/plugins/operators/transformers/print_string.py",
            python_callable_name:"main",
            op_kwargs:[
                {
                    name:"string_to_print",
                    pres_name:"String to print",
                    pres_desc:"String to be printed by the operator",
                    type:"string"
                }
            ]
        }
    }
)

The entire description of all parameters is described in Operator Metadata. For this example the OPERATOR_INPUT_PARAMETERS necessary are simply a string.

Strategy to Submit New Operators

To integrate a new operator into DAVE:

  1. Create a new branch:

    git checkout -b feature/new_operator

  2. Add the new operator files (.json and .py).

  3. Test the operator locally and in a staging environment.

  4. Commit and push the changes:

    git add example_operator.json example_operator.py
git commit -m "Added ExampleOperator"
git push origin feature/new_operator

  5. Create a Merge Request (MR):

    • Ensure all tests pass.

    • Request a review from the DAVE development team.

  6. Merge the MR once approved.

By following this structured approach, new operators can be seamlessly integrated into DAVE.

Useful links

Operator Metadata

Operator Scripts