Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[WIP] Fix issue #285 : save hive partitioned dataset using NativeExecutionEngine and DaskExecutionEngine #306

Merged
merged 18 commits into from
Apr 3, 2022
Merged

[WIP] Fix issue #285 : save hive partitioned dataset using NativeExecutionEngine and DaskExecutionEngine #306

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
18 commits
Select commit Hold shift + click to select a range
53d9e7d
Work in progress to fix issue 285 reported here https://github.com/fu…
LaurentErreca Feb 27, 2022
af81b59
Use option partition_on in Dask execution engine to write hive partit…
LaurentErreca Feb 28, 2022
36c17b8
Add handling for spark array type (#307)
WangCHX Mar 14, 2022
1b32865
adding ecosystem to README
kvnkho Mar 19, 2022
9a12cb6
adding ecosystem to README
kvnkho Mar 19, 2022
3e6cfda
downloads badge
kvnkho Mar 19, 2022
42daae4
merge conflict
kvnkho Mar 19, 2022
4419961
Merge pull request #309 from kvnkho/master
kvnkho Mar 19, 2022
f2676e0
Fugue plugin (#311)
goodwanghan Mar 24, 2022
d0c7d96
upgrading black version
kvnkho Apr 2, 2022
a53fc9c
fixing black version
kvnkho Apr 2, 2022
73d551b
Merge pull request #314 from kvnkho/black_version
kvnkho Apr 2, 2022
7678f31
Work in progress to fix issue 285 reported here https://github.com/fu…
LaurentErreca Feb 27, 2022
3f1d7e4
Use option partition_on in Dask execution engine to write hive partit…
LaurentErreca Feb 28, 2022
8af3bb0
Handle hive partitioning with Duckdb execution engine
LaurentErreca Apr 3, 2022
d4fa047
Merge commit
LaurentErreca Apr 3, 2022
a561717
Clean code with pylint
LaurentErreca Apr 3, 2022
b3e86b5
Use ArrowDataFrame(df.as_arrow()) instead of ArrowDataFrame(df.native…
LaurentErreca Apr 3, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 2 additions & 4 deletions fugue/execution/native_execution_engine.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -376,10 +376,8 @@ def save_df(
force_single: bool = False,
**kwargs: Any,
) -> None:
if not partition_spec.empty:
self.log.warning( # pragma: no cover
"partition_spec is not respected in %s.save_df", self
)
if not force_single and not partition_spec.empty:
kwargs['partition_cols'] = partition_spec.partition_by
self.fs.makedirs(os.path.dirname(path), recreate=True)
df = self.to_df(df)
save_df(df, path, format_hint=format_hint, mode=mode, fs=self.fs, **kwargs)
Expand Down
5 changes: 2 additions & 3 deletions fugue_dask/execution_engine.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -459,13 +459,12 @@ def save_df(
format_hint=format_hint,
mode=mode,
partition_spec=partition_spec,
force_single=force_single,
**kwargs,
)
else:
if not partition_spec.empty:
self.log.warning( # pragma: no cover
"partition_spec is not respected in %s.save_df", self
)
kwargs['partition_on'] = partition_spec.partition_by
self.fs.makedirs(os.path.dirname(path), recreate=True)
df = self.to_df(df)
save_df(df, path, format_hint=format_hint, mode=mode, fs=self.fs, **kwargs)
14 changes: 14 additions & 0 deletions fugue_test/builtin_suite.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1175,6 +1175,7 @@ def init_tmpdir(self, tmpdir):
def test_io(self):
path = os.path.join(self.tmpdir, "a")
path2 = os.path.join(self.tmpdir, "b.test.csv")
path3 = os.path.join(self.tmpdir, "c.partition")
with self.dag() as dag:
b = dag.df([[6, 1], [2, 7]], "c:int,a:long")
b.partition(num=3).save(path, fmt="parquet", single=True)
Expand All @@ -1185,6 +1186,19 @@ def test_io(self):
a.assert_eq(dag.df([[1, 6], [7, 2]], "a:long,c:int"))
a = dag.load(path2, header=True, columns="c:int,a:long")
a.assert_eq(dag.df([[6, 1], [2, 7]], "c:int,a:long"))
with self.dag() as dag:
b = dag.df([[6, 1], [2, 7]], "c:int,a:long")
b.partition(by='c').save(path3, fmt="parquet", single=False)
b.save(path2, header=True)
assert FileSystem().isdir(path3)
assert FileSystem().isdir(os.path.join(path3, 'c=6'))
assert FileSystem().isdir(os.path.join(path3, 'c=2'))
# TODO: in test below, once issue #288 is fixed, use dag.load instead of pd.read_parquet
pd.testing.assert_frame_equal(
pd.read_parquet(path3).sort_values('a').reset_index(drop=True),
pd.DataFrame({'c': pd.Categorical([6, 2]), 'a': [1, 7]}).reset_index(drop=True),
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we assert c is int type?

check_like=True
)

def test_save_and_use(self):
path = os.path.join(self.tmpdir, "a")
Expand Down