Daft connector
Daft is a distributed dataframe library for Python. Its Gravitino connector ships with Daft itself: it lists the catalogs, schemas and tables of a metalake, reads tables, and reads and writes files in filesets. The connector is for the Gravitino REST interface, so it needs the API host, not the Iceberg REST host.
For Iceberg tables, read through the Iceberg REST service instead, as in Iceberg tables; that path vends storage credentials scoped to each table.
Quick start
-
Install Daft with the Gravitino option.
Install Daftpip install "daft[gravitino]" -
Connect to the metalake as a local user.
{gravitino_host}is the published server API host; see Client access.Connect and read a tableimport os
from daft.catalog import Catalog
catalog = Catalog.from_gravitino(
endpoint="https://{gravitino_host}",
metalake_name="{metalake}",
auth_type="simple",
username="{username}",
password=os.environ["GRAVITINO_PASSWORD"],
)
print(catalog.list_tables("{catalog}.{schema}"))
df = catalog.get_table("{catalog}.{schema}.{table}").read()
df.show()
The caller needs USE_CATALOG and USE_SCHEMA, and SELECT_TABLE to read a table; see
Privileges.
Sign in
| Sign-in | Arguments | Sent as |
|---|---|---|
| Local user | auth_type="simple", username, password | HTTP Basic credentials |
| Identity provider | auth_type="oauth2", token | A bearer token |
Always pass a password or a token. With auth_type="simple" and no password, Daft sends only the
user name, which the server does not accept as a credential. Daft takes a token you already hold; it
does not request or refresh one from the identity provider.
Iceberg tables
Daft reads Iceberg tables through PyIceberg, which connects to the Iceberg REST service with vended credentials. Install PyIceberg alongside Daft:
pip install "daft[iceberg]"
import os
import daft
from pyiceberg.catalog import load_catalog
iceberg = load_catalog(
"lake",
**{
"type": "rest",
"uri": "https://{iceberg_host}/iceberg",
"warehouse": "{catalog}",
"token": os.environ["GRAVITINO_TOKEN"],
"header.X-Iceberg-Access-Delegation": "vended-credentials",
},
)
df = daft.read_iceberg(iceberg.load_table("{schema}.{table}"))
df.show()
See PyIceberg for the other ways PyIceberg signs in.
Filesets
Daft reads and writes fileset files with gvfs:// paths, in the form
gvfs://fileset/{catalog}/{schema}/{fileset}/{path}.
import os
import daft
from daft.io import IOConfig, GravitinoConfig
io_config = IOConfig(
gravitino=GravitinoConfig(
endpoint="https://{gravitino_host}",
metalake_name="{metalake}",
auth_type="simple",
username="{username}",
password=os.environ["GRAVITINO_PASSWORD"],
)
)
df = daft.read_parquet("gvfs://fileset/{catalog}/{schema}/{fileset}/**/*.parquet", io_config=io_config)
df.write_parquet("gvfs://fileset/{catalog}/{schema}/{fileset}/output/", io_config=io_config)
Writing through gvfs:// works for filesets on S3.
Storage credentials
The connector does not request vended credentials. To read a table it takes the storage keys from the catalog's properties, and the server returns those keys masked, so the connector cannot use them. Give the Python environment its own access to the storage, or read Iceberg tables through the Iceberg REST service as shown in Iceberg tables.
Related
- Daft Gravitino connector reference, from the Daft project
- Iceberg REST service
- Filesets
- How engines connect