Lakehouse Manager Dataset connector

Pokaż jako Markdown

In order to create a Connector and use it to interact with a Lakehouse Manager table, you can connect to the Default Dataset or to a Custom Dataset

Objective

Info

Using Dataset connector, you may query, insert, update or delete content programmatically.

Connect to the Lakehouse Manager

In order to create a Connector and use it to interact with a Lakehouse Manager table, you can connect to the Default Dataset or to a Custom Dataset.

Below, you will see the connection strings used for each type:

from forepaas.dwh import connect

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

After that you can use the cn_default.list() method to see the tables available in your Lakehouse Manager and then cn_default.select(...) or cn_default.query(...) to get the data from the table you want.

See the next section of this article for additional details.

Warning

Note that tables need to be loaded in the Lakehouse Manager before using the Connector.list() method and other Connector object methods. In other words, you have to create a table first in the Lakehouse Manager in order to use a table in the SDK.

Connector methods

list()

Get the list of tables in the dataset.

Output

Type	Description	Example
list	list of table names	`["chicago_calendar_full", "stations_rides"]`

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

Execute a SQL query on a compatible source, returns a dataframe (default), an iterable cursor, or a list of dict.

Input Parameters

Name	Type	Description	Example
sql	str	SQL query to execute	`select * from stations_rides`
limit	int	maximum number of results (-1: no limit)	-1
return_type	str	type of return	'dataframe', 'cursor' or 'dict'

Output

Type	Description	Example
mixed	results in the choosen type (dataframe, cursor, list[dict]

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

Extract data from a table with simple filters, returns a dataframe (default), an iterable cursor, or a list of dict.

Input Parameters

Name	Type	Description	Example
table_name	str	source table name	`stations_rides`
conds	dict	Dict of values to use as filters. `Keys` are attributes to filter with `values` as the filtering values. `List` produces a IN operator, otherwise produce a `=` operator	None / `{"date":["01/11/2013", "01/12/2013"], "station_name":"Harlem-Lake"}`
limit	int	maximum number of results. (-1: no limit)	-1
return_type	string	type of return	'dataframe', 'cursor' or 'dict'

Output

Type	Description	Example
mixed	results in the choosen type (dataframe, cursor, list[dict]

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

Performs an UPDATE SQL query through simple parameters and return the number of affected rows.

Input Parameters

Name	Type	Description	Example
table_name	str	name of the source table	`'stations_rides'`
set	dict	Dict of values to update. Key as the field name to update, values as the value to update.	`{“rides”:0}`
conds	dict	Dict of values to use as filters. `Keys` are attributes to filter with `values` as the filtering values. `List` produces a IN operator, otherwise produce a `=` operator	None / `{"date":["01/11/2013", "01/12/2013"], "station_name":"Harlem-Lake"}`
ignore	boolean	If `True`, performs an `UPDATE IGNORE` query. Otherwise `UPDATE` query.	`True`, `False`

Outputs

Type	Description	Example
int	Number of updated rows (if DBMS compatible)	42

delete(table_name, conds)

Performs a deletion based on a specific query with filtering conditions. Returns the number of deleted rows.
If you don’t need any filter, use the truncate method described below.

Input Parameters

Name	Type	Description	Example
table_name	str	name of the source table	`'stations_rides'`
conds	dict	Dict of values to use as filters. `Keys` are attributes to filter with `values` as the filtering values. `List` produces a IN operator, otherwise produce a `=` operator	`{"station_name":"Harlem-Lake"}`

Outputs

Type	Description	Example
int	Number of deleted rows (if DBMS compatible)	42

truncate(table_name)

Truncates (clears) all rows from a table.

Input Parameters

Name	Type	Description	Example
table_name	str	name of the source table	`'stations_rides'`

Outputs

Type	Description	Example
boolean	True if truncate succeeded. False if table does not exist. Raise Exception if anything else occurred.	`True`, `False`

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

Executes an insert query, adds on duplicate key update (ODKU) operation optionally, and returns statistics about the insertion of the batch.
If you activate the ODKU option, the query will automatically performs updates on existing rows instead of returning errors, regarding the existing primary key values in each row that you are inserting.

Warning

Please note that you MUST provide, for each row that you want to insert, at least the primary key of the destination table.

Info

We encourage you to use the module bulk_insert instead of this function, as documented in the Connector methods section of this page, because it has better management of data chunk and insertion.

Input Parameters

Name	Type	Description	Example
table_name	str	name of the source table	`'stations_rides'`
rows	dataframe	Dataframe to insert or update.	`pd.DataFrame([ {"station_id": 1, "station_name":"First Station"}, {"station_id":2, "name":"Second Station"}])`
odku	boolean	`False` = Don’t perform an ODKU; `True` = Performs an ODKU	`True`, `False`
returnWarnings	boolean	If True, return warnings aggregated and counted for each type of warning strings returned	True / False

Outputs

Type	Description	Example
dict	Statistics about the inserted batch.	`{"inserts":13, "skipped":10, "updates":3, "records": 30, "affected":4, "warnings": {"integer truncated":4}}`

drop_table(table_name)

drops the table from the query engine

Input Parameters

Name	Type	Description	Example
table_name	str	name of the source table	`'stations_rides'`

Outputs

Type	Description	Example
boolean	True if drop succeeded. False if table does not exist. Raise Exception if anything else occurred.	`True`, `False`

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

Return a table schema as a list of forepaas.dwh.attributes.AttributeSchema object

Input Parameters

Name	Type	Description	Example
table_name	str	name of the source table	`'stations_rides'`
catalog_name	str	name of the catalog. If `None`, the default connector's catalog is used.	`'default_dataset'`
schema_name	str	name of the schema. If `None` uniqueness of the table will be checked across all catalog's schemas.	`'default_schema'`
strict	bool	if true, will raise an error if a column has an unsupported type. Otherwise will just log a warning.	True / False

Output Parameters

Type	Description
list	AttributeSchema object can have 4 attributes: `name`, `data_type`, `nullable` & `default_value`

Example Output

[
    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)
]

LogicalObject Methods

In order to use the LogicalObject methods you will need to first import it using:

from forepaas.dwh.logical import LogicalObject

logical = LogicalObject() 
logical.list()

build(objects)

Launches a logical build in the Lakehouse Manager.

Input Parameters

Name	Type	Description	Example
objects	`list`	A list of logical objects to build.	`['user_data', 'product_metrics']`

Outputs

Type	Description
`int`	The status code of the build result.

Output Example

200 # Build successful

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

Creates a logical object based on an existing physical table.

Note: Currently works only with Trino and PostgreSQL catalogs.

Input Parameters

Name	Type	Description	Example
table	`str`	The name of the physical table.	`'raw_customer_data'`
dataset	`str`	The dataset name. Works only with Trino/PostgreSQL datasets. Default: `'default_dataset'`.	`'my_trino_dataset'`
kwargs	`dict`	Named arguments passed as JSON content to the API for additional configuration.	`{'owner': 'data_team', 'tags': ['customers', 'raw']}`

Outputs

Type	Description
`LogicalObject`	The created logical object.

get(name)

Retrieves a logical object from the Lakehouse Manager.

Input Parameters

Name	Type	Description	Example
name	`str`	The name of the logical object to retrieve.	`'sales_dashboard_view'`

Outputs

Type	Description
`LogicalObject`	The retrieved logical object.

list(dataset_name='default_dataset')

Lists all logical objects in the Lakehouse Manager.

Input Parameters

Name	Type	Description	Example
dataset_name	`str`	The dataset to filter logical objects by. Defaults to `"default_dataset"`.	`'default_dataset'`

Outputs

Type	Description
`list[LogicalObject]`	A list of logical objects.

remove(name)

Removes a logical object from the Lakehouse Manager.

Input Parameters

Name	Type	Description	Example
name	`str`	The name of the logical object to remove.	`'old_temp_table'`

Outputs

Type	Description
`object`	The status of the removal operation.

Output Example

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

Complementary notes

Due to lacks of metrics collected by PostgreSQL or Snowflake libraries, the statistics returned by Data Platform connectors are calculated based on limited factors.

When no error happens, inserts and skipped rows are calculated based on total rows before and after inserting, as well as the records (length of data to insert) provided. For example, if we try to insert 3 rows into a table of 5 rows, and the result is 7 rows in total, we consider 2 rows are inserted and 1 row skipped, while records is 3.
If error happens in a batch of data, the whole batch will be marked as warnings.
Note for Snowflake: statistics returned by insert_many() and insert_dataframe() wont have warnings calculated, all rows fail to insert or skipped will be marked as skipped

Go further

If you need training or technical assistance to implement our solutions, contact your sales representative or click on this link to get a quote and ask our Professional Services experts for a custom analysis of your project.

Ask questions, give your feedback and interact directly with the team building the Data Platform on the dedicated Discord channel.

If you need support with your OVHcloud services, create a request in our Help Centre.

Join our community of users.

Czy ta strona była pomocna?