Python: Add conversion from iceberg table scan to ray dataset #7148

JonasJ-ap · 2023-03-20T20:54:11Z

Added Feature

Add to_ray method to allow conversion from iceberg table scan to ray dataset

Example:

ray_dataset = table_test_null_nan_rewritten.scan().to_ray()
print(ray_dataset)

Output

Dataset(num_blocks=1, num_rows=3, schema={idx: int32, col_numeric: float})

jackye1995 · 2023-03-20T22:13:55Z

python/tests/test_integration.py

+
+
+@pytest.mark.integration
+def test_ray_dataset_nan(table_test_null_nan_rewritten: Table) -> None:


can we add some more tests?

Sure! I created a new table containing all the types in the test environment and used it to test the conversion from iceberg to ray dataset

python/pyiceberg/table/__init__.py

c21 · 2023-03-22T00:09:32Z

python/pyiceberg/table/__init__.py

+    def to_ray(self) -> ray.data.dataset.Dataset:
+        import ray
+
+        return ray.data.from_arrow(self.to_arrow())


just for my understanding, would self.to_arrow() returns a pa.Table has all rows from Iceberg table? So that would assume the table is very small to fit in the memory of Python client, right? What if the table is larger than the available memory?

Thank you for mentioning this. to_arrow() will convert all the rows and fields(columns) that are included in the current table scan to a pa.Table. Row filters or field selectors could be applied to the table scan to exclude unwanted rows or fields. e.g.

not_nan = table_test_null_nan.scan(row_filter=NotNaN("col_numeric"), selected_fields=("idx",)).to_arrow()

The huge table size will be an issue for to_arrow if the table scan ends up including the whole table (or too many rows). There is an open PR #7163 that may help in this case by adding a limit to the number of rows included in the table scan. Here is one example quoted from that PR:

limited_result = table_test_limit.scan(selected_fields=("idx",), limit=20).to_arrow()

In the future, I think there could be more discussions on converting large iceberg table to ray dataset. Please let me know if you have more questions/concerns on this.

Is this the same experience today for connections like duckdb?

I think so, since currently all the connections are based on to_arrow(). I draw the above conclusion based on the discussion here: #6505 (comment), #6505 (comment)

An out-of-memory error occurs when converting a large table to duckdb on an EC2 instance with insufficient memory.

Got it. I guess we need to address this issue across connections then. Thanks for the explanation!

Arrow also allows to_batches which will consume the source in batches, but not sure if Ray can leverage this. In the end, we want to leverage the PyArrow dataset, but this requires much deeper integration with Arrow. I'm digging into this, but no design yet. Maybe we can send a substrait plan? Open for discussion

It looks like self.to_arrow() will eventually call pq.read_table() to read the table into memory (with optionally row filtering and columns pruning) (to_arrow -> project_table -> _file_to_table -> pq.read_table).

Ray also leverages PyArrow dataset to read a variety of format (Parquet, CSV, etc) (API list). And internally Ray uses PyArrow to read data in batches and in parallel with many tasks.

This PR is a good starting point. We can further collaborate to add a read_iceberg() API in Ray, to implement a custom data source for Iceberg, so it can do the parallel read and be scalable.

…h in 20s

jackye1995

Looks good to me! @Fokko do you have any concerns in getting this integration in pyiceberg?

Fokko · 2023-03-22T19:01:04Z

python/pyproject.toml

 pyarrow = ["pyarrow"]
 pandas = ["pandas", "pyarrow"]
 duckdb = ["duckdb", "pyarrow"]
+ray = ["ray", "pyarrow", "pandas"]


Do we need pandas as well?

Sorry wondering what's this pyproject.toml used for? Is it the required dependency?

@Fokko Yes. The pandas seems to be a required dependency when converting from pa.Table to ray dataset.

Here is the trace when running table.scan().to_ray() without pandas:

2023-03-22 19:22:32,946 INFO worker.py:772 -- Task failed with retryable exception: TaskID(c8ef45ccd0112571ffffffffffffffffffffffff01000000). (_get_metadata pid=77891) Traceback (most recent call last): (_get_metadata pid=77891) File "python/ray/_raylet.pyx", line 857, in ray._raylet.execute_task (_get_metadata pid=77891) File "python/ray/_raylet.pyx", line 861, in ray._raylet.execute_task (_get_metadata pid=77891) File "/Users/jonasjiang/Library/Caches/pypoetry/virtualenvs/pyiceberg-oQQETec1-py3.9/lib/python3.9/site-packages/ray/data/read_api.py", line 1565, in _get_metadata (_get_metadata pid=77891) return BlockAccessor.for_block(table).get_metadata( (_get_metadata pid=77891) File "/Users/jonasjiang/Library/Caches/pypoetry/virtualenvs/pyiceberg-oQQETec1-py3.9/lib/python3.9/site-packages/ray/data/block.py", line 379, in for_block (_get_metadata pid=77891) import pandas (_get_metadata pid=77891) ModuleNotFoundError: No module named 'pandas'

Here is the reference for for_block method in ray:
https://github.com/ray-project/ray/blob/2e6de87ada84609fcd4f85bd16b9c2e090e921ae/python/ray/data/block.py#L379-L394

The import pandas cause this requirement.

@c21 The pyproject.toml tracks both the required and optional dependencies. Currently, ray is set as an optional dependency

ray = { version = ">=2.0.0,<=2.3.0", optional = true }

For example. If users want to convert the iceberg table stored in AWS Glue to a ray dataset, they can install the pyiceberg by

git clone https://github.com/apache/iceberg.git cd iceberg/python pip3 install -e ".[glue,ray]"

In this way, all the required dependencies for ray are installed, including ray, pyarrow, and pandas as specified in line 105

ref: https://py.iceberg.apache.org/

If this file represents required dependency, this change looks good to me.

Fokko

This looks great @JonasJ-ap! Could you update the docs as well? Add the ray option to the documentation.

JonasJ-ap · 2023-03-22T23:18:44Z

@Fokko Thanks for your review! I added the ray option and also an example to the documentation

JonasJ-ap · 2023-03-23T05:20:34Z

python/Makefile

 	docker-compose -f dev/docker-compose-integration.yml build
 	docker-compose -f dev/docker-compose-integration.yml up -d
-	sleep 20
+	sleep 30


According to #7163 (comment), #7163 (comment), it seems sometimes the iceberg-spark docker need more time to finish table creation. Extending the wait time to 30s to ensure the integration test can pass.

python/mkdocs/docs/api.md

jackye1995

Thanks for the work! I think Fokko's comment is also addressed, I will go ahead and merge when CI passes.

jackye1995 · 2023-03-23T16:38:41Z

Thanks for the contribution @JonasJ-ap , looking forward to more development and collaboration with the Ray community. Thanks @Fokko and @c21 for reviewing!

github-actions bot added the python label Mar 20, 2023

jackye1995 reviewed Mar 20, 2023

View reviewed changes

python/pyiceberg/table/__init__.py Outdated Show resolved Hide resolved

c21 reviewed Mar 22, 2023

View reviewed changes

JonasJ-ap added 5 commits March 21, 2023 21:49

add ray support

25214a6

revise the integration tests

a0838c6

update lock file

04c47cd

rename and add more tests

fa4ca76

reduce the number of rows in test table to make sure everything finis…

b813a85

…h in 20s

JonasJ-ap force-pushed the add_ray_dataset_support branch from 5f8014b to b813a85 Compare March 22, 2023 01:53

jackye1995 approved these changes Mar 22, 2023

View reviewed changes

Fokko reviewed Mar 22, 2023

View reviewed changes

add doc for ray

c3adbd8

JonasJ-ap added 2 commits March 23, 2023 01:13

fix format issue for doc

dfb2c7f

extends wait time for integration tests

54b00c6

JonasJ-ap commented Mar 23, 2023

View reviewed changes

jackye1995 reviewed Mar 23, 2023

View reviewed changes

python/mkdocs/docs/api.md Outdated Show resolved Hide resolved

revise doc

7f9dfa9

jackye1995 approved these changes Mar 23, 2023

View reviewed changes

jackye1995 merged commit afb40ff into apache:master Mar 23, 2023

Fokko mentioned this pull request Mar 23, 2023

Python: Add limit parameter to table scan #7163

Merged

jackye1995 mentioned this pull request Mar 31, 2023

Support read Iceberg dataset in Ray ray-project/deltacat#115

Closed



		@pytest.mark.integration
		def test_ray_dataset_nan(table_test_null_nan_rewritten: Table) -> None:

Python: Add conversion from iceberg table scan to ray dataset #7148

Python: Add conversion from iceberg table scan to ray dataset #7148

Uh oh!

Conversation

JonasJ-ap commented Mar 20, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Added Feature

Example:

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

JonasJ-ap Mar 22, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jackye1995 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

JonasJ-ap Mar 22, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Fokko left a comment

Choose a reason for hiding this comment

Uh oh!

JonasJ-ap commented Mar 22, 2023

Uh oh!

JonasJ-ap Mar 23, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

jackye1995 left a comment

Choose a reason for hiding this comment

Uh oh!

jackye1995 commented Mar 23, 2023

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

JonasJ-ap commented Mar 20, 2023 •

edited

Loading

JonasJ-ap Mar 22, 2023 •

edited

Loading

JonasJ-ap Mar 22, 2023 •

edited

Loading

JonasJ-ap Mar 23, 2023 •

edited

Loading