Creazione di un Service

Prima di continuare, se non lo hai già fatto, consulta la definizione di Service

La seguente figura mostra la struttura semplificata di un generico Service:

generic-service-structure

tutorial-label | Vedremo adesso passo passo come creare un semplice Service da zero fino alla sua registrazione sulla piattaforma. Esso calcolerà dei semplici indicatori statistici quali media e deviazione standard per un dataset in input, salvando i risultati su di un dataset di output.

Dataset di input e dataset di output verranno rispettivamente letto e scritto da/su S3.

Tale Service avrà quindi:

Una porta di input di tipo dataset
Una porta di output di tipo dataset
Un parametro di esecuzione configurabile

come schematizzato in figura:

to-be-created-service

Procediamo quindi:

Creando il programma nucleo contenente la logica di elaborazione dati
Buildando un'immagine Docker contenente tale programma
Caricando l'immagine Docker buildata su di un Docker Registry
Registrando il programma dockerizzato su ALIDA promuovendolo così a Service

1. Creazione Programma Nucleo

Porta di Input

service-with-single-input-dataset

Predisponiamo la Porta di Input per il nostro Service aggiungendo i seguenti argomenti linea di comando al programma nucleo:

--input-dataset
--input-columns
--input-dataset.s3_bucket
--input-dataset.s3_URL
--input-dataset.s3_ACCESS_KEY
--input-dataset.s3_SECRET_KEY
--input-dataset.s3_REGION

Ecco il codice di esempio per questi primi argomenti:

Tipo dati argomenti

I valori passati da ALIDA al programma nucleo saranno sempre stringhe o numeri, pertanto occorrerà convertire correttamente il valore al tipo dati desiderato lato codice.

Esempio Python

Per i booleani utilizzare str2bool
Per i JSON utilizzare str2json

import argparse
import pandas as pd
from minio import Minio
import os

parser = argparse.ArgumentParser(description="Basic Statistical Indicators")

# Argomenti CLI per la porta di input dataset
parser.add_argument('--input-dataset', 
  dest='input_dataset', 
  type=str, 
  required=True
)

parser.add_argument('--input-columns', 
  dest='input_columns', 
  type=str, 
  required=True
) 

parser.add_argument('--input-dataset.s3_bucket', 
  dest='input_dataset_s3_bucket', 
  type=str, 
  required=True
)

parser.add_argument('--input-dataset.s3_URL', 
  dest='input_dataset_s3_url', 
  type=str, 
  required=True
)

parser.add_argument('--input-dataset.s3_ACCESS_KEY', 
  dest='input_dataset_s3_access_key', 
  type=str, 
  required=True
)

parser.add_argument('--input-dataset.s3_SECRET_KEY', 
  dest='input_dataset_s3_secret_key', 
  type=str, 
  required=True
)

parser.add_argument('--input-dataset.s3_REGION',
 dest='input_dataset_s3_region',
 type=str,
 required=False
)


# ... see next step ...

Porta di Output

service-with-single-output-dataset

Predisponiamo ora la Porta di Output per il nostro Service aggiungendo i seguenti argomenti linea di comando al programma nucleo:

--output-dataset
--output-dataset.s3_bucket
--output-dataset.s3_URL
--output-dataset.s3_ACCESS_KEY
--output-dataset.s3_SECRET_KEY
--output-dataset.s3_REGION

Ecco il codice di esempio per questi ulteriori argomenti:

# ... omitted - see previous step ...

# Argomenti CLI per la porta di output dataset
parser.add_argument('--output-dataset', 
  dest='output_dataset', 
  type=str, 
  required=True
)

parser.add_argument('--output-dataset.s3_bucket', 
  dest='output_dataset_s3_bucket', 
  type=str, 
  required=True
)

parser.add_argument('--output-dataset.s3_URL', 
  dest='output_dataset_s3_url', 
  type=str, 
  required=True
)

parser.add_argument('--output-dataset.s3_ACCESS_KEY', 
  dest='output_dataset_s3_access_key', 
  type=str, 
  required=True
)

parser.add_argument('--output-dataset.s3_SECRET_KEY', 
  dest='output_dataset_s3_secret_key', 
  type=str, 
  required=True
)

parser.add_argument('--output-dataset.s3_REGION',
 dest='output_dataset_s3_region',
 type=str,
 required=False
)

# ... see next step ...

Parametro di esecuzione configurabili dall'utente

Predisponiamo adesso la porta per un parametro di esecuzione ausiliario. Questo permetterà all'utente di specificare l'indicatore statistico da calcolare lato UI.

setting-service-param-from-ui

Aggiungiamo quindi al programma nucleo il seguente argomento linea di comando:

--indicators-to-compute

Ecco il codice di esempio per questo ulteriore argomento:

# ... omitted - see previous steps ...

# Argomento CLI per il parametro di esecuzione configurabile dall'utente
parser.add_argument(
  '--indicators-to-compute', 
  dest='indicators_to_compute', 
  type=str, 
  required=True, 
  choices=['mean', 'standard_deviation', 'all_supported']
)

# parse_known_args permette di ignorare gli argomenti aggiuntivi iniettati da ALIDA
args, unknowns = parser.parse_known_args()

# ... see next step ...

Ecco quindi schematizzato il nostro programma nucleo con tutti gli argomenti desiderati:

Nota — parser permissivo obbligatorio

L'uso di parse_known_args() al posto di parse_args() è obbligatorio in tutti i Service ALIDA. ALIDA inietta sempre argomenti aggiuntivi a runtime — in particolare --go_manager.* e gli argomenti specifici della Datasource — che il programma nucleo non dichiara esplicitamente. Un parser rigido (parse_args()) causerebbe un crash immediato all'avvio del container.

to-be-created-service-with-core-program-args

Logica di esecuzione

Aggiunti tutti gli argomenti linea di comando, completiamo il programma con la logica di calcolo degli indicatori statistici. Di seguito il codice completo.

Codice programma nucleo finale

import argparse
import pandas as pd
from minio import Minio
import os

parser = argparse.ArgumentParser(description="Basic Statistical Indicators")

# Argomenti CLI per la porta di input dataset
parser.add_argument('--input-dataset', 
  dest='input_dataset', 
  type=str, 
  required=True
)

parser.add_argument('--input-columns', 
  dest='input_columns', 
  type=str, 
  required=True
) 

parser.add_argument('--input-dataset.s3_bucket', 
  dest='input_dataset_s3_bucket', 
  type=str, 
  required=True
)

parser.add_argument('--input-dataset.s3_URL', 
  dest='input_dataset_s3_url', 
  type=str, 
  required=True
)

parser.add_argument('--input-dataset.s3_ACCESS_KEY', 
  dest='input_dataset_s3_access_key', 
  type=str, 
  required=True
)

parser.add_argument('--input-dataset.s3_SECRET_KEY', 
  dest='input_dataset_s3_secret_key', 
  type=str, 
  required=True
)

parser.add_argument('--input-dataset.s3_REGION',
dest='input_dataset_s3_region',
type=str,
required=False
)

# Argomenti CLI per la porta di output dataset
parser.add_argument('--output-dataset', 
  dest='output_dataset', 
  type=str, 
  required=True
)

parser.add_argument('--output-dataset.s3_bucket', 
  dest='output_dataset_s3_bucket', 
  type=str, 
  required=True
)

parser.add_argument('--output-dataset.s3_URL', 
  dest='output_dataset_s3_url', 
  type=str, 
  required=True
)

parser.add_argument('--output-dataset.s3_ACCESS_KEY', 
  dest='output_dataset_s3_access_key', 
  type=str, 
  required=True
)

parser.add_argument('--output-dataset.s3_SECRET_KEY', 
  dest='output_dataset_s3_secret_key', 
  type=str, 
  required=True
)

parser.add_argument('--output-dataset.s3_REGION',
dest='output_dataset_s3_region',
type=str,
required=False
)

# Argomento CLI per il parametro di esecuzione configurabile dall'utente
parser.add_argument(
  '--indicators-to-compute', 
  dest='indicators_to_compute', 
  type=str, 
  required=True, 
  choices=['mean', 'standard_deviation', 'all_supported']
)

# parse_known_args permette di ignorare gli argomenti aggiuntivi iniettati da ALIDA
args, unknowns = parser.parse_known_args()

minio_client = Minio(
    args.input_dataset_s3_url.replace("http://", "").replace("https://", ""),
    access_key=args.input_dataset_s3_access_key,
    secret_key=args.input_dataset_s3_secret_key,
    region=args.input_dataset_s3_region,
    secure=False
)

objects = list(
    minio_client.list_objects(
        args.input_dataset_s3_bucket, 
        prefix=args.input_dataset, 
        recursive=True
    )
)

if not objects:
  raise FileNotFoundError("Nessun file trovato nel percorso S3 specificato.")

# Si assume un solo file CSV
csv_object = objects[0]
csv_filename = os.path.basename(csv_object.object_name)

"""
Il programma è progettato per leggere da S3, quindi è necessario
gestire la connessione all'object storage
    """
    connection_details = {
        'key': args.input_dataset_s3_access_key,
        'secret': args.input_dataset_s3_secret_key,
        'region': args.input_dataset_s3_region,
        'client_kwargs': {
            'endpoint_url': f'{args.input_dataset_s3_url}'
        }
    }

file_path = f"s3://{args.input_dataset_s3_bucket}/{args.input_dataset}/{csv_filename}"

try:
    # Lettura
    df = pd.read_csv(file_path, storage_options=connection_details, sep=None, engine='python')
    columns = args.input_columns
    if columns is not None and columns.strip() != '*':
        df_selected = df[[c.strip() for c in columns.split(",")]]
        df = df_selected.copy()
    numeric_cols_df = df.select_dtypes(include='number')

    # Calcolo
    operation = args.indicators_to_compute
    if operation == 'mean':
        result = numeric_cols_df.mean().to_frame(name='mean').T
        result["ResultType"] = "Mean"
    elif operation == 'standard_deviation':
        result = numeric_cols_df.std().to_frame(name='std').T
        result["ResultType"] = "Standard Deviation"
    elif operation == 'all_supported':
        mean_df = numeric_cols_df.mean().to_frame(name='mean').T
        mean_df["ResultType"] = "Mean"

        std_df = numeric_cols_df.std().to_frame(name='std').T
        std_df["ResultType"] = "Standard Deviation"
        result = pd.concat([mean_df, std_df])
    else:
        raise ValueError(
            "Operation must be 'mean', " \
            "'standard_deviation', or "
            "'all_supported'." \
            "")

    # Salvataggio
    csv_filename = 'basic_statistical_indicators_results.csv'
    file_path = f"s3://{args.output_dataset_s3_bucket}/{args.output_dataset}/{csv_filename}"

    result.to_csv(
      file_path,
      index=False,
      storage_options={
        'key': args.output_dataset_s3_access_key,
        'secret': args.output_dataset_s3_secret_key,
        'region': args.output_dataset_s3_region,
        'client_kwargs': {
            'endpoint_url': f'{args.output_dataset_s3_url}'
        }
      }
    )

except Exception as e:
    print(f"Error: {e}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187

2/3. Build e Caricamento Immagine Docker

Procediamo adesso alla creazione dell'immagine Docker incapsulante il programma nucleo.

Nota

In fase di esecuzione del Service, ALIDA istanzierà tale immagine passandole gli argomenti linea di comando valorizzati. Il container Docker istanziato dovrà quindi passare gli argomenti al programma nucleo. Affinché ciò avvenga, l'immagine Docker dovrà avere un opportuno ENTRYPOINT.

Per il nostro Service ecco il Dockerfile (notare l'entrypoint):

dockerfile

FROM python:3.13.7-alpine3.22

RUN pip install pandas==2.3.2 \
                minio==7.2.18 \
                fsspec==2025.9.0 \
                s3fs==2025.9.0

COPY . . 

ENTRYPOINT ["python", "main.py", "$@"]

Gli argomenti valorizzati passati da ALIDA al container Docker verranno sostituiti a $@.

Altri linguaggi

Nel caso di altri linguaggi, l'entrypoint assumerà forme del tipo:

ENTRYPOINT ["./main", "$@"]
ENTRYPOINT ["java", "Main", "$@"]
ENTRYPOINT ["julia", "main.jl", "$@"]
etc ...

Procediamo quindi al build e push dell'immagine Docker eseguendo i seguenti comandi.

Il registry Docker può essere di due tipi:

Utilizzare un registry pubblicamente accessibile
Utilizzare un registry accessibile dalla piattaforma ALIDA, nel cui caso occorre contattare l'amministratore.

bash

docker build -t <dominio-registry-docker>:<porta>/<percorso-immagine>:<tag> .

docker login <dominio-registry-docker>:<porta> -u <username_docker>

docker push <dominio-registry-docker>:<porta>/<percorso-immagine>:<tag>

4. Registrazione del Service creato

Con l'immagine caricata sul registry, è possibile registrare il corrispondente Service su ALIDA rendendolo così disponibile a catalogo per la creazione di Workflow.

Variabili d'ambiente iniettate da ALIDA

A runtime, ALIDA inietta automaticamente le seguenti variabili d'ambiente in ogni container Service:

Variabile	Sempre presente	Semantica
`SERVICE_ID`	Sì	Identificativo del Service; obbligatorio nei messaggi Workflow Media
`EXECUTION_ID`	Sì	Identificativo dell'esecuzione corrente; obbligatorio nei messaggi Workflow Media
`HASH_TOKEN`	Sì	Token di autenticazione; obbligatorio nei messaggi Workflow Media
`ARGO_NODE_ID`	Sì	Identificativo del nodo Argo che esegue il container
`FATHER_ID`	Sì	Identificativo del contesto di esecuzione padre (distinto da `EXECUTION_ID`)
`BDA_ID`	Sì	Identificativo dell'istanza del Workflow
`NAMESPACE`	Sì	Namespace dell'istanza ALIDA (sottodominio in `{NAMESPACE}.alidalab.it`)
`MLFLOW_TRACKING_URI`	Solo se MLflow abilitato	URI del server MLflow; verificarne la raggiungibilità prima dell'uso

Le Service Property di tipo Static sono iniettate come variabili d'ambiente aggiuntive.

Integrazione MLflow

Per i Service che tracciano esperimenti su MLflow, ALIDA impone alcuni vincoli specifici affinché le run siano visibili e navigabili dall'UI della piattaforma.

Experiment naming

Il nome dell'esperimento MLflow deve essere impostato a partire da FATHER_ID, non da EXECUTION_ID:

python

import os
if os.environ.get("FATHER_ID") is not None:
    os.environ["MLFLOW_EXPERIMENT_NAME"] = os.environ.get("FATHER_ID")

Run naming

Il nome della run MLflow deve essere impostato tramite il tag mlflow.runName con il valore di ARGO_NODE_ID:

python

if os.environ.get("ARGO_NODE_ID") is not None:
    mlflow.set_tag("mlflow.runName", os.environ.get("ARGO_NODE_ID"))

Tag obbligatorio per Service che producono un modello

Ogni Service che produce un modello e traccia su MLflow deve impostare il tag alida-model-identifier, che identifica univocamente il modello nel contesto dell'esecuzione ALIDA:

python

bda_id = os.environ.get("BDA_ID")
service_id = os.environ.get("SERVICE_ID")
execution_id = os.environ.get("EXECUTION_ID")
if all([bda_id, service_id, execution_id]):
    mlflow.set_tag("alida-model-identifier", f"{bda_id}_{service_id}_{execution_id}")

Naming del modello nel Model Registry

Il nome del modello registrato nel Model Registry MLflow deve seguire la convenzione model-<output_model_id>, dove output_model_id è il valore dell'argomento --output-model.id iniettato da ALIDA a runtime:

python

model_name = "model-" + str(args.output_model_id)
mlflow.sklearn.log_model(pipeline, "model", registered_model_name=model_name)

Nota

MLFLOW_TRACKING_URI è presente solo se MLflow è abilitato nell'istanza ALIDA. Verificarne la raggiungibilità prima di aprire una run.

Accediamo al form di registrazione Service come segue:

Accediamo alla pagina di gestione dei Service dal menù laterale
Clicchiamo su + Register Service in alto a destra

sidebar-with-services-item-highlighted

Si aprirà il form di registrazione Service

service-registration-form

Qui impostiamo per prima cosa i metadati di base per il Service:

Name: basic-statistical-indicators
Version: 1.0.0
URL: docker://<dominio-tuo-registry-docker>:<porta>/<percorso-immagine>:<tag>

Versionamento dei Service

Il campo Version permette di impostare una versione per il Service che si sta registrando.

Notare come sia possibile definire diverse versioni dello stesso Service aventi la stessa immagine Docker. Questo permette di creare più versioni dello stesso Service diverse l'una dall'altra per esempio per descrizione, documentazione o categorie associate.

Tutte le versioni di un Service compariranno a catalogo e conseguentemente sulla palette del Workflow Designer.

Fatto ciò, passiamo alle Service Properties. Esse comunicano ad ALIDA quali sono le Porte I/O e i Parametri di configurazione del Service che essa dovrà valorizzare prima dell'esecuzione.

Aggiungiamo la prima Service Property al nostro Service per la porta di input: --input-dataset

Cliccare su + Add Property per aprire il form di inserimento:

service-registration-add-property-form

E riempiamo il form come segue --input-dataset:

Key: input-dataset
Description: lasciare vuota
Type: Application Property
Mandatory: ticked
Invisible: ticked
Value Type: String
Default Value: lasciare vuoto
Data Type: Input Data
Streaming: unticked

e clicchiamo su Confirm

Allo stesso modo creiamo le restanti Service Property:

Per --input-columns:
- Key: input-columns
- Description: lasciare vuota
- Type: Application Property
- Mandatory: ticked
- Invisible: ticked
- Value Type: String
- Default Value: ANY
- Data Type: Input Data
- Streaming: unticked
Per --output-dataset:
- Key: output-dataset
- Description: lasciare vuota
- Type: Application Property
- Mandatory: ticked
- Invisible: ticked
- Value Type: String
- Default Value: lasciare vuoto
- Data Type: Output Data
- Streaming: unticked
Per --indicators-to-compute:
- Key: indicators-to-compute
- Description: lasciare vuota
- Type: Application Property
- Mandatory: ticked
- Invisible: unticked
- Value Type: String
- Default Value: lasciare vuoto
- Data Type: Input Data
- Streaming: unticked

Alla fine avremo le seguenti Service Property:

new-service-tutorial-service-properties-summary

Registriamo il Service completo di properties cliccando su Save in basso a destra:

Una volta salvato comparirà la pagina di dettaglio del Service

just-registered-service-details-page

Il Service è adesso disponibile a catalogo e può essere utilizzato per creare un Workflow

Prossimi passi

Utilizza il Service appena creato in un Workflow (come visto in Primi Passi)
Visita le sezioni:

Layout

Agenti ALIDA

Creazione di un Service

1. Creazione Programma Nucleo

Porta di Input

Porta di Output

Parametro di esecuzione configurabili dall'utente

Logica di esecuzione

2/3. Build e Caricamento Immagine Docker

4. Registrazione del Service creato

Integrazione MLflow

Prossimi passi

Layout

Creazione di un Service ​

1. Creazione Programma Nucleo ​

Porta di Input ​

Porta di Output ​

Parametro di esecuzione configurabili dall'utente ​

Logica di esecuzione ​

2/3. Build e Caricamento Immagine Docker ​

4. Registrazione del Service creato ​

Integrazione MLflow ​

Prossimi passi ​

Creazione di un Service

1. Creazione Programma Nucleo

Porta di Input

Porta di Output

Parametro di esecuzione configurabili dall'utente

Logica di esecuzione

2/3. Build e Caricamento Immagine Docker

4. Registrazione del Service creato

Integrazione MLflow

Prossimi passi