dask-trino 0.2.0

Dask + Trino integration

dask-trino

This connector is an experimental integration between Dask and Trino. It only works using the spooling protocol introduced in trino version 466.

Installation

dask-trino can be installed with pip:

pip install dask-trino

trino configuration

The spooling protocol is currently disabled by default when you deploy trino. The trino docs outline how to enable it.

One extra consideration for the dask-trino connector is the expiry of segments on object storage. By default, when a client acknowledges a segment, the segment is deleted from object storage. We recommend disabling this and letting segments be pruned by trino. This can be achieved by adding this setting to the spooling-manager.properties configuration file:

fs.segment.explicit-ack=false

We recommend this because when working with a dask dataframe, the data may be requested multiple times. If a segment is deleted after the first request, the client will not be able to read from the segment again.

Usage

dask-trino provides read_trino and to_trino methods for parallel IO from trino with Dask.

>>> from dask_trino import read_trino
>>> example_query = '''
...    SELECT *
...    FROM tpch.sf1.customer
... '''
>>> ddf = read_trino(
...     query=example_query,
...     connection_kwargs={
...         "user": "...",
...         "password": "...",
...     },
... )

Note that the to_trino method generates bulk INSERT statements for each partition of the dataframe. There is no other option to do bulk insert of data into trino at this time.

>>> from dask_trino import to_trino
>>> to_trino(
...     ddf,
...     name="my_table",
...     connection_kwargs={
...         "user": "...",
...         "password": "...",
...     },
... )