read_hudi#

ray.data.read_hudi(table_uri: str, *, query_type: str = 'snapshot', filters: List[Tuple[str, str, str]] | None = None, hudi_options: Dict[str, str] | None = None, storage_options: Dict[str, str] | None = None, num_cpus: float | None = None, num_gpus: float | None = None, memory: float | None = None, concurrency: int | None = None, override_num_blocks: int | None = None, label_selector: Dict[str, str] | None = None, fallback_strategy: List[Dict[str, Any]] | None = None, max_calls: int | None = None, resources: Dict[str, float] | None = None, accelerator_type: str | None = None, runtime_env: Dict[str, Any] | None = None, ray_remote_args: Dict[str, Any] | None = None) → Dataset[source]#

Create a Dataset from an Apache Hudi table.

Examples

>>> import ray
>>> ds = ray.data.read_hudi(
...     table_uri="/hudi/trips",
...     query_type="snapshot",
...     filters=[("city", "=", "san_francisco")],
... )

>>> ds = ray.data.read_hudi(
...     table_uri="/hudi/trips",
...     query_type="incremental",
...     hudi_options={
...         "hoodie.read.file_group.start_timestamp": "20230101123456789",
...         "hoodie.read.file_group.end_timestamp": "20230201123456789",
...     },
... )
Parameters:

  • table_uri (str) – The URI of the Hudi table to read from. Local file paths, S3, and GCS are supported.

  • query_type (str) – The Hudi query type to use. Supported values are snapshot and incremental.

  • filters (List[Tuple[str, str, str]] | None) – Optional list of filters to apply to the Hudi table when the query_type is snapshot. Each filter is a tuple of the form (column_name, operator, value). The operator can be one of "=", "!=", "<", "<=", ">", ">=". Currently, only filters on partition columns will be effective.

  • hudi_options (Dict[str, str] | None) – A dictionary of Hudi options to pass to the Hudi reader.

  • storage_options (Dict[str, str] | None) – Extra options that make sense for a particular storage connection. This is used to store connection parameters like credentials, endpoint, etc. See more explanation here.

  • num_cpus (float | None) – The number of CPUs to reserve for each parallel read worker.

  • num_gpus (float | None) – The number of GPUs to reserve for each parallel read worker. For example, specify num_gpus=1 to request 1 GPU for each parallel read worker.

  • memory (float | None) – The heap memory in bytes to reserve for each parallel read worker.

  • concurrency (int | None) – The maximum number of Ray tasks to run concurrently. Set this to control number of tasks to run concurrently. This doesn’t change the total number of tasks run or the total number of output blocks. By default, concurrency is dynamically decided based on the available resources.

  • override_num_blocks (int | None) – Override the number of output blocks from all read tasks. By default, the number of output blocks is dynamically decided based on input data size and available resources. You shouldn’t manually set this value in most cases.

  • label_selector (Dict[str, str] | None) – Labels required on the node where each 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.

  • max_calls (int | None) – The maximum number of read tasks a worker runs before exiting.

  • resources (Dict[str, float] | None) – Custom resources to reserve for each read task, expressed as a mapping from resource name to quantity.

  • accelerator_type (str | None) – The accelerator type required for each read task.

  • runtime_env (Dict[str, Any] | None) – The runtime environment to use for each read task.

  • ray_remote_args (Dict[str, Any] | None) – Additional options passed to ray.remote() for each read task. This argument is deprecated and will be removed in Ray 2.64. Use the named remote parameters instead.

Returns:

A Dataset producing records read from the Hudi table.

Return type:

Dataset

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

On this page