Skip to content

Fix write_dataframe_async parallel write overwriting #1310

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

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Fix write_dataframe_async parallel write overwriting #1310

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
8 changes: 6 additions & 2 deletions TM1py/Services/CellService.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@
abbreviate_mdx,
build_cellset_from_pandas_dataframe,
build_csv_from_cellset_dict,
build_dataframe_aggregate_intersections,
build_dataframe_from_csv,
build_mdx_and_values_from_cellset,
build_mdx_from_cellset,
Expand Down Expand Up @@ -1068,7 +1069,7 @@ def write_dataframe_async(
slice_size_of_dataframe: int = 250_000,
max_workers: int = 8,
dimensions: Iterable[str] = None,
increment: bool = False,
increment: bool = True,
sandbox_name: str = None,
deactivate_transaction_log: bool = False,
reactivate_transaction_log: bool = False,
Expand All @@ -1083,7 +1084,7 @@ def write_dataframe_async(
:param slice_size_of_dataframe: Number of rows for each DataFrame slice, e.g. 10000
:param max_workers: Max number of threads, e.g. 14
:param dimensions:
:param increment: increment or update cell values. Defaults to False.
:param increment: increment or update cell values. Defaults to True.
:param sandbox_name: name of the sandbox or None
:param deactivate_transaction_log:
:param reactivate_transaction_log:
Expand Down Expand Up @@ -1127,6 +1128,9 @@ async def _write_async(df: "pd.DataFrame"):

return failures

if increment:
data = build_dataframe_aggregate_intersections(data, sum_numeric_duplicates=True)

exceptions = asyncio.run(_write_async(data))
if not exceptions:
return
Expand Down
17 changes: 17 additions & 0 deletions TM1py/Utils/Utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -937,6 +937,23 @@ def build_cellset_from_pandas_dataframe(

:return: a CaseAndSpaceInsensitiveTuplesDict
"""
df = build_dataframe_aggregate_intersections(df, sum_numeric_duplicates)

cellset = CaseAndSpaceInsensitiveTuplesDict(
dict(zip(df.iloc[:, :-1].itertuples(index=False, name=None), df.iloc[:, -1].values))
)
return cellset


@require_pandas
def build_dataframe_aggregate_intersections(df: "pd.DataFrame", sum_numeric_duplicates: bool = True) -> "pd.DataFrame":
"""

param sum_numeric_duplicates: Aggregate numerical values for duplicated intersections
param df: A Dataframe, with dimension-column mapping in correct order.

:return: A Dataframe
"""
if isinstance(df.index, pd.MultiIndex):
df.reset_index(inplace=True)

Expand Down