You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by GitBox <gi...@apache.org> on 2022/09/20 07:54:44 UTC
[GitHub] [iceberg] Fokko opened a new issue, #5800: Integrate pyiceberg with Dask
Fokko opened a new issue, #5800:
URL: https://github.com/apache/iceberg/issues/5800
### Feature Request / Improvement
It would be awesome to integrate pyiceberg with Dask to allow reading in data from iceberg tables into Dask.
### Query engine
Other
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Integrate pyiceberg with Dask [iceberg]
Posted by "TomAugspurger (via GitHub)" <gi...@apache.org>.
TomAugspurger commented on issue #5800:
URL: https://github.com/apache/iceberg/issues/5800#issuecomment-3387561907
Thanks @mikeoconnor0308. If that's just using public APIs from pyiceberg then I think we'd be able to add a `dask.dataframe.read_iceberg` function to Dask.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Integrate pyiceberg with Dask [iceberg]
Posted by "theearthwanderer (via GitHub)" <gi...@apache.org>.
theearthwanderer commented on issue #5800:
URL: https://github.com/apache/iceberg/issues/5800#issuecomment-2088835891
Hello, is there any update on this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Integrate pyiceberg with Dask [iceberg]
Posted by "mikeoconnor0308 (via GitHub)" <gi...@apache.org>.
mikeoconnor0308 commented on issue #5800:
URL: https://github.com/apache/iceberg/issues/5800#issuecomment-3386193114
I'd also like to see some movement on a native Dask solution.
Here's my take on @grobgl 's implementation that accounts for changes to pyIceberg APIs and doesn't pass io things around.
Same limitations - no row group information or division support. I haven't tested it at large scale yet.
```python
class IcebergDaskIO(DataFrameIOFunction):
_metadata_location: str
_projected_schema: Schema
_row_filter: Union[str, BooleanExpression]
_case_sensitive: bool
def __init__(
self,
metadata_location: str,
projected_schema: Schema,
row_filter: Union[str, BooleanExpression] = ALWAYS_TRUE,
case_sensitive: bool = True,
):
super().__init__()
self._metadata_location = metadata_location
self._projected_schema = projected_schema
self._row_filter = row_filter
self._case_sensitive = case_sensitive
@classmethod
def from_scan(cls, scan: DataScan, metadata_location: str) -> "IcebergDaskIO":
if scan.limit:
raise ValueError("IcebergDaskIO does not support scans with a limit")
return cls(
# TODO feels like we should be able to get this from scan, but scan.metadata.location isn't the full path?
metadata_location=metadata_location,
projected_schema=scan.projection(),
row_filter=scan.row_filter,
case_sensitive=scan.case_sensitive,
)
# region DataFrameIOFunction interface
@property
def columns(self) -> List[str]:
return list(self._projected_schema.column_names)
@property
def empty_table(self) -> pd.DataFrame:
return (
schema_to_pyarrow(self._projected_schema)
.empty_table()
.to_pandas(date_as_object=False, types_mapper=pd.ArrowDtype)
)
def project_columns(self, columns: Sequence[str]) -> "IcebergDaskIO":
cols = list(columns)
if cols == self.columns:
return self
new_schema = self._projected_schema.select(*cols)
return IcebergDaskIO(
metadata_location=self._metadata_location,
projected_schema=new_schema,
row_filter=self._row_filter,
case_sensitive=self._case_sensitive,
)
def __call__(self, task: FileScanTask, *, columns: list[str] | None = None) -> pd.DataFrame:
if columns is not None:
schema = self._projected_schema.select(*columns)
else:
schema = self._projected_schema
tbl = self._load_table()
arrow_tbl = ArrowScan(
tbl.metadata,
tbl.io,
schema,
self._row_filter,
self._case_sensitive,
).to_table([task])
if arrow_tbl is None or arrow_tbl.num_rows == 0:
return self.empty_table
return arrow_tbl.to_pandas(types_mapper=pd.ArrowDtype)
# endregion
@lru_cache(maxsize=1)
def _load_table(self) -> StaticTable:
return StaticTable.from_metadata(self._metadata_location)
```
Sample usage:
```python
table = catalog.load_table("my_namespace.my_table")
scan = table.scan()
io_func = IcebergDaskIO.from_scan(scan, metadata_location=table.metadata_location)
tasks = list(scan.plan_files())
return dd.from_map(
io_func,
tasks,
meta=io_func.empty_table,
enforce_metadata=True,
)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Integrate pyiceberg with Dask [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on issue #5800:
URL: https://github.com/apache/iceberg/issues/5800#issuecomment-2594191567
This issue has been automatically marked as stale because it has been open for 180 days with no activity. It will be closed in next 14 days if no further activity occurs. To permanently prevent this issue from being considered stale, add the label 'not-stale', but commenting on the issue is preferred when possible.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] TomAugspurger commented on issue #5800: Integrate pyiceberg with Dask
Posted by "TomAugspurger (via GitHub)" <gi...@apache.org>.
TomAugspurger commented on issue #5800:
URL: https://github.com/apache/iceberg/issues/5800#issuecomment-1443857629
Here's a rough version of a `DataScane` to Dask DataFrame.
```python
from pyiceberg.catalog import load_catalog
from pyiceberg.expressions import GreaterThanOrEqual
def _file_to_pandas(fs, task, bound_row_filter, projected_schema, projected_field_ids, case_sensitive):
from pyiceberg.io.pyarrow import _file_to_table
return _file_to_table(fs, task, bound_row_filter, projected_schema, projected_field_ids, case_sensitive).to_pandas()
def to_dask_dataframe(sc):
"""Convert a DataScane to a Dask DataFrame"""
from pyiceberg.io.pyarrow import (
PyArrowFileIO, bind, extract_field_ids, schema_to_pyarrow, MapType, ListType,
)
import pyarrow as pa
import dask
import dask.dataframe as dd
# arguments
tasks = list(sc.plan_files())
table = sc.table
row_filter = sc.row_filter
projected_schema = sc.projection()
case_sensitive = sc.case_sensitive
# stuff stolen from to_arrow()
if isinstance(table.io, PyArrowFileIO):
scheme, _ = PyArrowFileIO.parse_location(table.location())
fs = table.io.get_fs(scheme)
else:
raise ValueError(f"Expected PyArrowFileIO, got: {table.io}")
bound_row_filter = bind(table.schema(), row_filter, case_sensitive=case_sensitive)
projected_field_ids = {
id for id in projected_schema.field_ids if not isinstance(projected_schema.find_type(id), (MapType, ListType))
}.union(extract_field_ids(bound_row_filter))
# build the Dask DataFrame
schema = schema_to_pyarrow(projected_schema)
names = [x.name for x in projected_schema.fields]
meta = pa.table([[]] * len(schema.names), schema=schema).to_pandas()
# TODO: ensure deterministic
token = dask.base.tokenize(fs, bound_row_filter, projected_schema, projected_field_ids, case_sensitive)
name = f'from-iceberg-{token}'
dsk = {
(name, i): (
_file_to_pandas, fs, task, bound_row_filter, projected_schema, projected_field_ids, case_sensitive
)
for i, task in enumerate(tasks)
}
divisions = [None] * len(dsk)
df = dd.DataFrame(dsk, name, meta, divisions)
return df
```
It seems to work, but I haven't tested it beyond a basic `df.head()`. A couple notes:
1. This returns a Dask DataFrame with a single Dask partition per "scan file" in `.scan_files()`. Which is maybe equal to the number of parquet files?
2. Currently, `divisions` is set to None, which is sub-optimal. (xref https://docs.dask.org/en/stable/dataframe-design.html?highlight=divisions#partitions, https://docs.dask.org/en/stable/dataframe-parquet.html?highlight=divisions#calculating-divisions). I'm seeing some stuff in `DataFile` that might help with setting those properly.
3. I've never used (py)iceberg before, but I was pleasantly surprised with how straightforward this was. This is copy-pasting bits and pieces out of `pyiceberg.io.pyarrow`. With a proper refactor, this is even fewer net new lines of code.
4. On the Dask side, we would be interested in making this more sophisticated, to allow operations on the Dask DataFrame to affect the original scan (https://github.com/dask/dask/issues/9970 and linked issues)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] TomAugspurger commented on issue #5800: Integrate pyiceberg with Dask
Posted by "TomAugspurger (via GitHub)" <gi...@apache.org>.
TomAugspurger commented on issue #5800:
URL: https://github.com/apache/iceberg/issues/5800#issuecomment-1553478936
> Generally, should this be part of the Dask library instead of PyIceberg?
I was wondering that too. I think either would be sensible, but I'd lean slightly towards putting the implementation in Dask (and pyiceberg could add a `.to_dask_dataframe` to mirror `to_pandas` if desired).
That said, the current implementation does use a private `_file_to_table`. I think pyiceberg would need to add a public method that achieves that before Dask could use it. And given the similarity to `pyiceberg.io.pyarrow`, maybe we could start with some refactoring to add some methods used by both the `pyarrow` and `dask.dataframe` implementations?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Integrate pyiceberg with Dask [iceberg]
Posted by "andoni-garcia-fgp (via GitHub)" <gi...@apache.org>.
andoni-garcia-fgp commented on issue #5800:
URL: https://github.com/apache/iceberg/issues/5800#issuecomment-2661089695
@Fokko I see you marked this as complete under the milestone PyIceberg 0.6.0 release. Is that accurate? I currently use a variant of grobgl's `DataFrameIOFunction` from above, but only for reading iceberg partitions into Dask frames. I would love if there is native Dask read/write solution. Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Integrate pyiceberg with Dask [iceberg]
Posted by "TomAugspurger (via GitHub)" <gi...@apache.org>.
TomAugspurger commented on issue #5800:
URL: https://github.com/apache/iceberg/issues/5800#issuecomment-2599332424
I think this is still waiting for a decision on https://github.com/apache/iceberg/issues/5800#issuecomment-1987287800
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] martindurant commented on issue #5800: Integrate pyiceberg with Dask
Posted by GitBox <gi...@apache.org>.
martindurant commented on issue #5800:
URL: https://github.com/apache/iceberg/issues/5800#issuecomment-1303714276
cc https://github.com/martindurant/daskberg/issues/1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] Fokko commented on issue #5800: Integrate pyiceberg with Dask
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on issue #5800:
URL: https://github.com/apache/iceberg/issues/5800#issuecomment-1617724107
PyIceberg 0.4.0 has been released today, and should fix the pickling issues! 👍🏻
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Integrate pyiceberg with Dask [iceberg]
Posted by "sam-goodwin (via GitHub)" <gi...@apache.org>.
sam-goodwin commented on issue #5800:
URL: https://github.com/apache/iceberg/issues/5800#issuecomment-1987174551
What's the latest on this issue? Also keen to write to Iceberg directly from Dask.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Integrate pyiceberg with Dask [iceberg]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko closed issue #5800: Integrate pyiceberg with Dask
URL: https://github.com/apache/iceberg/issues/5800
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Integrate pyiceberg with Dask [iceberg]
Posted by "TomAugspurger (via GitHub)" <gi...@apache.org>.
TomAugspurger commented on issue #5800:
URL: https://github.com/apache/iceberg/issues/5800#issuecomment-1987287800
I think the main decision point was around where this should live: dask or pyiceberg.
I don't see the private `_file_to_table` in https://github.com/apache/iceberg-python/blob/main/pyiceberg/io/pyarrow.py anymore. If we can write the snippets provided in this issue using just public pyiceberg APIs, I think that a PR to Dask would be reasonable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] grobgl commented on issue #5800: Integrate pyiceberg with Dask
Posted by "grobgl (via GitHub)" <gi...@apache.org>.
grobgl commented on issue #5800:
URL: https://github.com/apache/iceberg/issues/5800#issuecomment-1553073273
I'm keen to push this forward. @TomAugspurger's implementation works in single-threaded mode but fails in a distributed scenario due to current lack of pickle support (I raised a separate issue #7644).
Extending Tom's approach, this is a solution which utilises Dask's `from_map` and `DataFramIOFunction` which allows us to pass projected columns to the Parquet reader:
```python
class IcebergFunctionWrapper(DataFrameIOFunction):
def __init__(
self,
fs: FileSystem,
bound_row_filter: BooleanExpression,
projected_schema: Schema,
case_sensitive: bool,
):
self._fs = fs
self._bound_row_filter = bound_row_filter
self._projected_schema = projected_schema
self._case_sensitive = case_sensitive
self._projected_field_ids = {
id for id in projected_schema.field_ids
if not isinstance(projected_schema.find_type(id), (MapType, ListType))
}.union(extract_field_ids(bound_row_filter))
super().__init__()
@property
def columns(self) -> List[str]:
self._projected_schema.column_names
@property
def empty_table(self) -> pd.DataFrame:
return schema_to_pyarrow(self._projected_schema).empty_table().to_pandas(date_as_object=False)
def project_columns(self, columns: Sequence[str]) -> 'IcebergFunctionWrapper':
if list(columns) == self.columns:
return self
return IcebergFunctionWrapper(
self._fs,
self._bound_row_filter,
self._projected_schema.select(*columns),
self._case_sensitive,
)
def __call__(self, task: FileScanTask) -> pd.DataFrame:
table = _file_to_table(
self._fs,
task,
self._bound_row_filter,
self._projected_schema,
self._projected_field_ids,
self._case_sensitive,
0, # no limit support yet
)
if table is None:
return self.empty_table
return table.to_pandas(date_as_object=False)
def to_dask_dataframe(scan: DataScan) -> dd.DataFrame:
tasks = scan.plan_files()
table = scan.table
row_filter = scan.row_filter
projected_schema = scan.projection()
case_sensitive = scan.case_sensitive
scheme, _ = PyArrowFileIO.parse_location(table.location())
if isinstance(table.io, PyArrowFileIO):
fs = table.io.get_fs(scheme)
else:
try:
from pyiceberg.io.fsspec import FsspecFileIO
if isinstance(table.io, FsspecFileIO):
fs = PyFileSystem(FSSpecHandler(table.io.get_fs(scheme)))
else:
raise ValueError(f"Expected PyArrowFileIO or FsspecFileIO, got: {table.io}")
except ModuleNotFoundError as e:
# When FsSpec is not installed
raise ValueError(f"Expected PyArrowFileIO or FsspecFileIO, got: {table.io}") from e
bound_row_filter = bind(table.schema(), row_filter, case_sensitive=case_sensitive)
io_func = IcebergFunctionWrapper(fs, bound_row_filter, projected_schema, case_sensitive)
return dd.from_map(
io_func,
tasks,
meta=io_func.empty_table,
enforce_metadata=False,
)
```
I'm also looking into adding `divisions` support and row-group-level parallelisation to this.
Generally, should this be part of the Dask library instead of PyIceberg?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Integrate pyiceberg with Dask [iceberg]
Posted by "mikeoconnor0308 (via GitHub)" <gi...@apache.org>.
mikeoconnor0308 commented on issue #5800:
URL: https://github.com/apache/iceberg/issues/5800#issuecomment-3389284794
Yes, it just uses public APIs.
However, from experimenting with it I think the lack of row-group information is pretty suboptimal, I ended up reading in some very chunky partitions into dask.dataframe.
I think it should be possible to perform the metadata lookups to get row group information, but from poking around PyIceberg APIs I think it may be end up leading to the use of lower-level details which would result in high coupling.
For example, it would make things tidier if `DataScan` or `ArrowScan` could produce a plan of batches/row groups.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Integrate pyiceberg with Dask [iceberg]
Posted by "martindurant (via GitHub)" <gi...@apache.org>.
martindurant commented on issue #5800:
URL: https://github.com/apache/iceberg/issues/5800#issuecomment-3392044507
@mikeoconnor0308 : if you really want low level, you can of course bypass iceberg altogether. I had a go at this a good while ago ( https://github.com/martindurant/daskberg ) and was pretty successful for as much of the feature set as I had time to try.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org