read_hive#

Read a HiveServer2 table or trusted SQL query into a Dataset.

Install the optional impyla package on the driver and Ray workers. Configure authentication, transport, and TLS certificate verification in connection_factory. GSSAPI also requires impyla[kerberos] and Kerberos credentials on those nodes. The real-server integration test covers PLAIN with HiveServer2 in server mode NONE. It doesn’t cover NOSASL or GSSAPI.

Note

Query reads require an explicit Arrow schema. DataSourceV2 planning needs it on the driver before the read worker executes the query. See issue #66840 for optional schema inference from the same query execution. Table reads select every column. If a table has unsupported column types, use query to select supported columns and provide a matching schema.

Each Dataset execution runs at most one HiveServer2 data query in one Ray task. Ray doesn’t retry failed reads. With the default error tolerance, a read error or worker loss fails that Dataset execution. This API preserves DataContext.max_errored_blocks. If you allow block errors, execution can skip failed output and return an incomplete result. This scanner doesn’t push Dataset filters, projections, or Dataset.limit() into HiveServer2. The limit argument adds a SQL LIMIT for table reads. The HiveServer2 reader ignores override_num_blocks with a warning to preserve streaming. Output blocks follow Ray’s normal block-sizing policy. This parameter doesn’t parallelize the HiveServer2 query.

The reader attempts to cancel the HiveServer2 operation and close its cursor and connection when reading ends. Closing a Dataset iterator can force Ray to stop the read worker before Python cleanup runs. Immediate operation cancellation or explicit connection cleanup isn’t guaranteed.

Examples

Read a table or a query result. These examples assume analytics.events has exactly two columns, id BIGINT and name STRING, in that order. The SELECT * example assumes HiveServer2 returns events.id and events.name as result labels. Match the labels returned by your HiveServer2 instance.

import os

import pyarrow as pa
import ray

def create_connection():
    from impala.dbapi import connect

    return connect(
        host="hive.example.com",
        port=10000,
        auth_mechanism="PLAIN",
        user="reader",
        password=os.environ["HIVE_NONE_PLACEHOLDER"],
        retries=1,
    )

table_ds = ray.data.read_hive(
    "analytics.events", connection_factory=create_connection
)
query_ds = ray.data.read_hive(
    query="SELECT * FROM analytics.events",
    schema=pa.schema(
        [("events.id", pa.int64()), ("events.name", pa.string())]
    ),
    connection_factory=create_connection,
)

# Use explicit SQL aliases to choose unqualified result labels.
aliased_query_ds = ray.data.read_hive(
    query=(
        "SELECT id AS event_id, name AS event_name "
        "FROM analytics.events"
    ),
    schema=pa.schema(
        [("event_id", pa.int64()), ("event_name", pa.string())]
    ),
    connection_factory=create_connection,
)

Parameters:

table (str | None) – The Hive table to read, optionally qualified as database.table. Specify exactly one of table and query.
connection_factory (Callable[[], Any]) – A serializable function that takes no arguments and returns a new Impyla HiveServer2 connection on each call. Ray calls it on the driver for table metadata and on the read worker for data. Query reads with an explicit schema only call it on the read worker. Ray closes each returned connection. Configure authentication, TLS verification, and client retries in the factory. Read credentials inside the factory to avoid capturing them in its serialized state. Dependencies, credentials, and certificate files must be available wherever the factory runs.
query (str | None) – A trusted, row-producing SQL query. Specify exactly one of table and query. The query is sent to HiveServer2 as given and must return a result set.
schema (pyarrow.Schema | None) – The Arrow schema for a query read. Column names must match the HiveServer2 result labels, case-insensitively and in order. Hive may prefix SELECT * result labels with the table name or alias. Use those labels in the schema or alias the columns in your query. Table reads infer their schema from HiveServer2 metadata; schema is only supported for query reads. Field types must use a supported Arrow mapping. Hive TIMESTAMP maps to pa.timestamp("us") without a time zone; sub-microsecond precision is truncated.
user (str | None) – The HiveServer2 session user or proxy user passed to connection.cursor(user=user). If omitted, Impyla uses the operating system username. Configure the authentication user separately in connection_factory.
limit (int | None) – The maximum number of rows to read from a table. 0 skips the data query. The limit is sent to HiveServer2 as SQL. This argument isn’t supported for query reads.
num_cpus (float | None) – The number of CPUs to reserve for the read task.
memory (float | None) – The heap memory in bytes to reserve for the read task.
resources (Dict[str, float] | None) – Custom resources to reserve for the read task, expressed as a mapping from resource name to quantity.
label_selector (Dict[str, str] | None) – Labels required on the node where the read task runs.
fallback_strategy (List[Dict[str, Any]] | None) – Alternative label requirements that Ray tries in order if label_selector can’t be satisfied.
runtime_env (Dict[str, Any] | None) – The runtime environment to use for the read task.
override_num_blocks (int | None) – An output block-count hint. The HiveServer2 reader ignores this hint and logs a warning when you supply a value. Output blocks follow Ray’s normal block-sizing policy. This parameter doesn’t add HiveServer2 query parallelism.

Returns:

A Dataset containing the HiveServer2 read result.

Return type:

Dataset

PublicAPI (alpha): This API is in alpha and may change before becoming stable.