Skip to content

Read from HDFS

Remote queries read HDFS paths and Iceberg tables on HDFS with the pure-Rust hdfs_native client, so no JVM is needed. This support is experimental and unstable.

Set up

Iceberg metadata needs the pyiceberg extra and the hdfs-native and pyiceberg Python packages in the worker environment, which you install like the other Python packages. Direct data access needs neither. The HDFS client reads core-site.xml and hdfs-site.xml from $HADOOP_CONF_DIR. With Kerberos, every worker needs the dynamic library libgssapi-krb5-2 and a Kerberos ticket cache on local disk.

worker:
  extras:
    hdfs:
      enabled: true

For Iceberg metadata, also set worker.extras.pyiceberg.enabled to true. Mount the Hadoop configuration directory into the workers with worker.deployment.volumes and worker.deployment.runtimeContainer.volumeMounts, and set HADOOP_CONF_DIR to the mount path in worker.deployment.runtimeContainer.env. See the HDFS section of the Helm chart documentation.

[worker.extras]
hdfs.enabled = true

For Iceberg metadata, also set pyiceberg.enabled = true and install the package with both extras: uv pip install "polars-nanook-compute[hdfs,pyiceberg]". Place the Hadoop configuration directory on every worker machine and set HADOOP_CONF_DIR to it in the service's environment. See the Bare-metal config file reference.

File scans

Remote queries scan HDFS paths with the standard open-source Polars API, for example:

import polars as pl
import polars_nanook as pn

ctx: pn.ClusterContext

src = "hdfs://<NAMENODE>:8020/data/foods1.parquet"

result = (
    pl.scan_parquet(src)
    .remote(context=ctx)
    .collect()
)

No storage_options are required. Any you pass go to the HDFS client.

Iceberg tables with metadata on HDFS

To read or write an Iceberg table with its metadata on HDFS, pass its metadata path or a table object.

The storage_options must include:

  • the py-io-impl key with the value shown below
  • the hdfs.host and optionally hdfs.port keys, set to the location per the pyiceberg spec. The host and port in the URI are ignored.

With a known metadata path:

import polars as pl
import polars_nanook as pn

ctx: pn.ClusterContext

storage_options = {
    "py-io-impl": "pyiceberg.io.fsspec.FsspecFileIO",
    "hdfs.host": "<NAMENODE>",
    "hdfs.port": "8020"
}

metadata_path = (
    "hdfs://<NAMENODE>:8020/warehouse/db/test/metadata/"
    "00001-<UUID>.metadata.json"
)

result = (
    pl.scan_iceberg(metadata_path, storage_options=storage_options)
    .remote(context=ctx)
    .collect()
)

Similarly, a table object can be used with a sink:

import polars_nanook as pn

ctx: pn.ClusterContext

# table: a pyiceberg Table object; df: a local DataFrame to append
storage_options = {
    "py-io-impl": "pyiceberg.io.fsspec.FsspecFileIO",
    "hdfs.host": "<NAMENODE>",
    "hdfs.port": "8020"
}

(
    df.lazy()
    .remote(context=ctx)
    .sink_iceberg(table, storage_options=storage_options, mode="append")
    .await_result()
)