Connecteur Dataset du Lakehouse Manager

Voir en Markdown

Afin de créer un connecteur et l'utiliser pour interagir avec une table du Lakehouse Manager, vous pouvez vous connecter au Default Dataset ou à un Custom Dataset

Objectif

Info

En utilisant le connecteur Dataset, vous pouvez interroger, insérer, mettre à jour ou supprimer du contenu de manière programmatique.

Se connecter au Lakehouse Manager

Afin de créer un connecteur et l'utiliser pour interagir avec une table du Lakehouse Manager, vous pouvez vous connecter au Default Dataset ou à un Custom Dataset.

Ci-dessous, vous trouverez les chaînes de connexion utilisées pour chaque type :

from forepaas.dwh import connect

cn_default = connect("dwh/default_dataset/")
cn_custom = connect("dwh/custom_dataset_name/")

Après cela, vous pouvez utiliser la méthode cn_default.list() pour voir les tables disponibles dans votre Lakehouse Manager, puis cn_default.select(...) ou cn_default.query(...) pour récupérer les données de la table que vous souhaitez.

Consultez la section suivante de cet article pour plus de détails.

Warning

Notez que les tables doivent être chargées dans le Lakehouse Manager avant d'utiliser la méthode Connector.list() et les autres méthodes de l'objet Connector. Autrement dit, vous devez d'abord créer une table dans le Lakehouse Manager afin de pouvoir utiliser une table dans le SDK.

Méthodes du connecteur

list()

Récupère la liste des tables du dataset.

Sortie

Type	Description	Exemple
list	liste des noms de table	`["chicago_calendar_full", "stations_rides"]`

query(sql, limit=-1, return_type='dataframe')

Exécute une requête SQL sur une source compatible, renvoie un dataframe (par défaut), un curseur itérable, ou une liste de dict.

Paramètres d'entrée

Nom	Type	Description	Exemple
sql	str	Requête SQL à exécuter	`select * from stations_rides`
limit	int	nombre maximum de résultats (-1 : pas de limite)	-1
return_type	str	type de retour	'dataframe', 'cursor' ou 'dict'

Sortie

Type	Description	Exemple
mixed	résultats dans le type choisi (dataframe, cursor, list[dict]

select(table_name, conds={}, limit=-1, return_type='dataframe' )

Extrait des données d'une table avec des filtres simples, renvoie un dataframe (par défaut), un curseur itérable, ou une liste de dict.

Paramètres d'entrée

Nom	Type	Description	Exemple
table_name	str	nom de la table source	`stations_rides`
conds	dict	Dict de valeurs à utiliser comme filtres. Les `Keys` sont les attributs à filtrer avec les `values` comme valeurs de filtrage. `List` produit un opérateur IN, sinon produit un opérateur `=`	None / `{"date":["01/11/2013", "01/12/2013"], "station_name":"Harlem-Lake"}`
limit	int	nombre maximum de résultats. (-1 : pas de limite)	-1
return_type	string	type de retour	'dataframe', 'cursor' ou 'dict'

Sortie

Type	Description	Exemple
mixed	résultats dans le type choisi (dataframe, cursor, list[dict]

update(table_name, set, conds=None, ignore=False)

Effectue une requête SQL UPDATE via des paramètres simples et renvoie le nombre de lignes affectées.

Paramètres d'entrée

Nom	Type	Description	Exemple
table_name	str	nom de la table source	`'stations_rides'`
set	dict	Dict de valeurs à mettre à jour. Key comme nom du champ à mettre à jour, values comme valeur à mettre à jour.	`{“rides”:0}`
conds	dict	Dict de valeurs à utiliser comme filtres. Les `Keys` sont les attributs à filtrer avec les `values` comme valeurs de filtrage. `List` produit un opérateur IN, sinon produit un opérateur `=`	None / `{"date":["01/11/2013", "01/12/2013"], "station_name":"Harlem-Lake"}`
ignore	boolean	Si `True`, effectue une requête `UPDATE IGNORE`. Sinon, une requête `UPDATE`.	`True`, `False`

Sorties

Type	Description	Exemple
int	Nombre de lignes mises à jour (si compatible SGBD)	42

delete(table_name, conds)

Effectue une suppression basée sur une requête spécifique avec des conditions de filtrage. Renvoie le nombre de lignes supprimées.
Si vous n'avez besoin d'aucun filtre, utilisez la méthode truncate décrite ci-dessous.

Paramètres d'entrée

Nom	Type	Description	Exemple
table_name	str	nom de la table source	`'stations_rides'`
conds	dict	Dict de valeurs à utiliser comme filtres. Les `Keys` sont les attributs à filtrer avec les `values` comme valeurs de filtrage. `List` produit un opérateur IN, sinon produit un opérateur `=`	`{"station_name":"Harlem-Lake"}`

Sorties

Type	Description	Exemple
int	Nombre de lignes supprimées (si compatible SGBD)	42

truncate(table_name)

Tronque (vide) toutes les lignes d'une table.

Paramètres d'entrée

Nom	Type	Description	Exemple
table_name	str	nom de la table source	`'stations_rides'`

Sorties

Type	Description	Exemple
boolean	True si le truncate a réussi. False si la table n'existe pas. Lève une exception si tout autre problème survient.	`True`, `False`

insert(table_name, rows, odku=0, returnWarnings=False)

Exécute une requête d'insertion, ajoute optionnellement une opération on duplicate key update (ODKU), et renvoie des statistiques sur l'insertion du lot.
Si vous activez l'option ODKU, la requête effectuera automatiquement des mises à jour sur les lignes existantes au lieu de renvoyer des erreurs, en fonction des valeurs de clé primaire existantes dans chaque ligne que vous insérez.

Warning

Veuillez noter que vous DEVEZ fournir, pour chaque ligne que vous souhaitez insérer, au moins la clé primaire de la table de destination.

Info

Nous vous encourageons à utiliser le module bulk_insert plutôt que cette fonction, comme documenté dans la section Méthodes du connecteur de cette page, car il gère mieux les blocs de données et l'insertion.

Paramètres d'entrée

Nom	Type	Description	Exemple
table_name	str	nom de la table source	`'stations_rides'`
rows	dataframe	Dataframe à insérer ou mettre à jour.	`pd.DataFrame([ {"station_id": 1, "station_name":"First Station"}, {"station_id":2, "name":"Second Station"}])`
odku	boolean	`False` = N'effectue pas d'ODKU ; `True` = Effectue un ODKU	`True`, `False`
returnWarnings	boolean	Si True, renvoie les avertissements agrégés et comptés pour chaque type de chaîne d'avertissement renvoyée	True / False

Sorties

Type	Description	Exemple
dict	Statistiques sur le lot inséré.	`{"inserts":13, "skipped":10, "updates":3, "records": 30, "affected":4, "warnings": {"integer truncated":4}}`

drop_table(table_name)

supprime la table du moteur de requêtes

Paramètres d'entrée

Nom	Type	Description	Exemple
table_name	str	nom de la table source	`'stations_rides'`

Sorties

Type	Description	Exemple
boolean	True si le drop a réussi. False si la table n'existe pas. Lève une exception si tout autre problème survient.	`True`, `False`

get_table_schema(table_name, catalog_name, schema_name, strict=True)

Renvoie un schéma de table sous forme d'une liste d'objets forepaas.dwh.attributes.AttributeSchema

Paramètres d'entrée

Nom	Type	Description	Exemple
table_name	str	nom de la table source	`'stations_rides'`
catalog_name	str	nom du catalogue. Si `None`, le catalogue par défaut du connecteur est utilisé.	`'default_dataset'`
schema_name	str	nom du schéma. Si `None`, l'unicité de la table sera vérifiée dans tous les schémas du catalogue.	`'default_schema'`
strict	bool	si true, lèvera une erreur si une colonne a un type non pris en charge. Sinon, se contentera de logger un avertissement.	True / False

Paramètres de sortie

Type	Description
list	L'objet AttributeSchema peut avoir 4 attributs : `name`, `data_type`, `nullable` et `default_value`

Exemple de sortie

[
    AttributeSchema(name='id', data_type=DataType.INT, nullable=False),
    AttributeSchema(name='name', data_type=DataType.VARCHAR, nullable=True),
    AttributeSchema(name='email', data_type=DataType.VARCHAR, nullable=False, default_value=''),
    AttributeSchema(name='created_at', data_type=DataType.TIMESTAMP, nullable=False)
]

Méthodes LogicalObject

Afin d'utiliser les méthodes LogicalObject, vous devrez d'abord l'importer avec :

from forepaas.dwh.logical import LogicalObject

logical = LogicalObject() 
logical.list()

build(objects)

Lance une construction logique dans le Lakehouse Manager.

Paramètres d'entrée

Nom	Type	Description	Exemple
objects	`list`	Une liste d'objets logiques à construire.	`['user_data', 'product_metrics']`

Sorties

Type	Description
`int`	Le code de statut du résultat de la construction.

Exemple de sortie

200 # Build réussi

create_from_physical(table, dataset='default_dataset', **kwargs)

Crée un objet logique basé sur une table physique existante.

Note : fonctionne actuellement uniquement avec les catalogues Trino et PostgreSQL.

Paramètres d'entrée

Nom	Type	Description	Exemple
table	`str`	Le nom de la table physique.	`'raw_customer_data'`
dataset	`str`	Le nom du dataset. Fonctionne uniquement avec les datasets Trino/PostgreSQL. Par défaut : `'default_dataset'`.	`'my_trino_dataset'`
kwargs	`dict`	Arguments nommés transmis sous forme de contenu JSON à l'API pour une configuration supplémentaire.	`{'owner': 'data_team', 'tags': ['customers', 'raw']}`

Sorties

Type	Description
`LogicalObject`	L'objet logique créé.

get(name)

Récupère un objet logique depuis le Lakehouse Manager.

Paramètres d'entrée

Nom	Type	Description	Exemple
name	`str`	Le nom de l'objet logique à récupérer.	`'sales_dashboard_view'`

Sorties

Type	Description
`LogicalObject`	L'objet logique récupéré.

list(dataset_name='default_dataset')

Liste tous les objets logiques du Lakehouse Manager.

Paramètres d'entrée

Nom	Type	Description	Exemple
dataset_name	`str`	Le dataset par lequel filtrer les objets logiques. Par défaut : `"default_dataset"`.	`'default_dataset'`

Sorties

Type	Description
`list[LogicalObject]`	Une liste d'objets logiques.

remove(name)

Supprime un objet logique du Lakehouse Manager.

Paramètres d'entrée

Nom	Type	Description	Exemple
name	`str`	Le nom de l'objet logique à supprimer.	`'old_temp_table'`

Sorties

Type	Description
`object`	Le statut de l'opération de suppression.

Exemple de sortie

{'success': True, 'info': None, '_id': '690325793a04befefec45bf1', 'uid': None}

Notes complémentaires

En raison du manque de métriques collectées par les bibliothèques PostgreSQL ou Snowflake, les statistiques renvoyées par les connecteurs Data Platform sont calculées à partir de facteurs limités.

Lorsqu'aucune erreur ne se produit, les lignes inserts et skipped sont calculées à partir du total de lignes avant et après l'insertion, ainsi que des records (nombre de données à insérer) fournis. Par exemple, si nous essayons d'insérer 3 lignes dans une table de 5 lignes, et que le résultat est 7 lignes au total, nous considérons que 2 lignes ont été inserted et 1 ligne skipped, tandis que records vaut 3.
Si une erreur se produit dans un lot de données, le lot entier sera marqué comme warnings.
Remarque pour Snowflake : les statistiques renvoyées par insert_many() et insert_dataframe() n'auront pas de warnings calculés, toutes les lignes qui échouent à l'insertion ou sont ignorées seront marquées comme skipped

Aller plus loin

Si vous avez besoin d'une formation ou d'une assistance technique pour la mise en oeuvre de nos solutions, contactez votre commercial ou cliquez sur ce lien pour obtenir un devis et demander une analyse personnalisée de votre projet à nos experts de l’équipe Professional Services.

Posez vos questions, faites-nous part de vos commentaires et interagissez directement avec l’équipe qui développe la Data Platform sur le canal Discord dédié.

Si vous avez besoin d'une assistance concernant vos services OVHcloud, créez une demande depuis notre centre d'aide.

Rejoignez notre communauté d'utilisateurs.

Cette page vous a-t-elle aidé ?