[Python] Python Dask try deduplicate index

I merge two Dask data frames

def merge_with_aggregated_4(trans_ddf, agg_ddf):
# First join condition: Adjust based on minutes
trans_ddf["base_hour"] = trans_ddf["column4"].dt.floor("h")
trans_ddf['adjusted_hour'] = trans_ddf['column4'].apply(adjust_hour_row, meta=('column4', 'datetime64[ns]'))

# First join
merged_ddf = trans_ddf.merge(agg_ddf, left_on="adjusted_hour", right_on="hour", how="left", suffixes=("", "_joined"))

# Identify rows that were not joined and remove columns belong to agg_ddf
unmatched_ddf = merged_ddf[merged_ddf["mean_column1"].isna()].drop(
columns=[col for col in agg_ddf.columns if col != "hour"]
)

# Secondary join based only on the original hour
unmatched_ddf = unmatched_ddf.merge(agg_ddf, left_on="base_hour", right_on="hour", how="left", suffixes=("", "_fallback"))

# Fill missing values using the second join
for col in agg_ddf.columns:
if col != "hour":
merged_ddf[col] = merged_ddf[col].fillna(unmatched_ddf[col])

return merged_ddf


And in test I print result index

result_ddf = merge_with_aggregated_4(trans_ddf, agg_ddf)

print(f' result index: {result_ddf.index.compute()}')


And it is duplicated result index: Index([2, 3, 7, 2, 0, 1, 4, 5, 6, 8, 9, 10, 0, 1], dtype='int64') This leads to problems in future when I try to do compute

I need deduplicate index.

Continue reading...