Esportare dati in ambienti ML esterni
Questo documento illustra come condividere un set di dati di formazione preparato creato con Data Distiller in una posizione di archiviazione cloud che l’ambiente di apprendimento può leggere per l’apprendimento e il punteggio del modello. L'esempio esporta il set di dati di formazione in Data Landing Zone (DLZ). È possibile modificare la destinazione di archiviazione in base alle esigenze per lavorare con l'ambiente di apprendimento automatico.
Il servizio Flusso per destinazioni viene utilizzato per completare la pipeline delle funzionalità inviando un set di dati di funzionalità calcolate in un percorso di archiviazione cloud appropriato.
Creare la connessione di origine create-source-connection
La connessione di origine è responsabile della configurazione della connessione al set di dati Adobe Experience Platform in modo che il flusso risultante sappia esattamente dove cercare i dati e in quale formato.
from aepp import flowservice
flow_conn = flowservice.FlowService()
training_dataset_id = <YOUR_TRAINING_DATASET_ID>
source_res = flow_conn.createSourceConnectionDataLake(
name=f"[CMLE] Featurized Dataset source connection created by {username}",
dataset_ids=[training_dataset_id],
format="parquet"
)
source_connection_id = source_res["id"]
Creare la connessione di destinazione create-target-connection
La connessione di destinazione è responsabile della connessione al file system di destinazione. A tal fine è necessario innanzitutto creare una connessione di base all’account di archiviazione cloud (la Data Landing Zone in questo esempio) e quindi una connessione di destinazione a un percorso di file specifico con opzioni di compressione e formato specificate.
Le destinazioni di archiviazione cloud disponibili sono identificate ciascuna da un ID della specifica di connessione:
connection_spec_id = "10440537-2a7b-4583-ac39-ed38d4b848e8"
base_connection_res = flow_conn.createConnection(data={
"name": "Base Connection to DLZ created by",
"auth": None,
"connectionSpec": {
"id": connection_spec_id,
"version": "1.0"
}
})
base_connection_id = base_connection_res["id"]
target_res = flow_conn.createTargetConnection(
data={
"name": "Data Landing Zone target connection",
"baseConnectionId": base_connection_id,
"params": {
"mode": "Server-to-server",
"compression": config.get("Cloud", "compression_type"),
"datasetFileType": config.get("Cloud", "data_format"),
"path": config.get("Cloud", "export_path")
},
"connectionSpec": {
"id": connection_spec_id,
"version": "1.0"
}
}
)
target_connection_id = target_res["id"]
Creare il flusso di dati create-data-flow
Il passaggio finale consiste nel creare un flusso di dati tra il set di dati specificato nella connessione di origine e il percorso del file di destinazione specificato nella connessione di destinazione.
Ogni tipo di archiviazione cloud disponibile è identificato da un ID della specifica di flusso:
Il codice seguente crea un flusso di dati con una pianificazione impostata per iniziare molto in futuro. Questo consente di attivare flussi ad hoc durante lo sviluppo del modello. Una volta creato un modello addestrato, è possibile aggiornare la pianificazione del flusso di dati per condividere il set di dati della feature sulla pianificazione desiderata.
import time
on_schedule = False
if on_schedule:
schedule_params = {
"interval": 3,
"timeUnit": "hour",
"startTime": int(time.time())
}
else:
schedule_params = {
"interval": 1,
"timeUnit": "day",
"startTime": int(time.time() + 60*60*24*365) # Start the schedule far in the future
}
flow_spec_id = "cd2fc47e-e838-4f38-a581-8fff2f99b63a"
flow_obj = {
"name": "Flow for Feature Dataset to DLZ",
"flowSpec": {
"id": flow_spec_id,
"version": "1.0"
},
"sourceConnectionIds": [
source_connection_id
],
"targetConnectionIds": [
target_connection_id
],
"transformations": [],
"scheduleParams": schedule_params
}
flow_res = flow_conn.createFlow(
obj = flow_obj,
flow_spec_id = flow_spec_id
)
dataflow_id = flow_res["id"]
Una volta creato il flusso di dati, ora puoi attivare un’esecuzione di flusso ad hoc per condividere il set di dati della funzione su richiesta:
from aepp import connector
connector = connector.AdobeRequest(
config_object=aepp.config.config_object,
header=aepp.config.header,
loggingEnabled=False,
logger=None,
)
endpoint = aepp.config.endpoints["global"] + "/data/core/activation/disflowprovider/adhocrun"
payload = {
"activationInfo": {
"destinations": [
{
"flowId": dataflow_id,
"datasets": [
{"id": created_dataset_id}
]
}
]
}
}
connector.header.update({"Accept":"application/vnd.adobe.adhoc.dataset.activation+json; version=1"})
activation_res = connector.postData(endpoint=endpoint, data=payload)
activation_res
Condivisione semplificata nella Data Landing Zone
Per condividere un set di dati nella Data Landing Zone in modo più semplice, la libreria aepp fornisce una funzione exportDatasetToDataLandingZone che esegue i passaggi precedenti in una singola chiamata di funzione:
from aepp import exportDatasetToDataLandingZone
export = exportDatasetToDataLandingZone.ExportDatasetToDataLandingZone()
dataflow_id = export.createDataFlowRunIfNotExists(
dataset_id = created_dataset_id,
data_format = data_format,
export_path= export_path,
compression_type = compression_type,
on_schedule = False,
config_path = config_path,
entity_name = "Flow for Featurized Dataset to DLZ"
)
Questo codice crea la connessione di origine, la connessione di destinazione e il flusso di dati in base ai parametri forniti ed esegue un’esecuzione ad hoc del flusso di dati in un singolo passaggio.