(
df_to_join: dd.DataFrame,
all_join_keys: List[str],
timestamp_field: str,
created_timestamp_column: str,
entity_df_event_timestamp_col: str,
)
| 1184 | |
| 1185 | |
| 1186 | def _drop_duplicates( |
| 1187 | df_to_join: dd.DataFrame, |
| 1188 | all_join_keys: List[str], |
| 1189 | timestamp_field: str, |
| 1190 | created_timestamp_column: str, |
| 1191 | entity_df_event_timestamp_col: str, |
| 1192 | ) -> dd.DataFrame: |
| 1193 | column_order = df_to_join.columns |
| 1194 | |
| 1195 | # try-catch block is added to deal with this issue https://github.com/dask/dask/issues/8939. |
| 1196 | # TODO(kevjumba): remove try catch when fix is merged upstream in Dask. |
| 1197 | try: |
| 1198 | if created_timestamp_column: |
| 1199 | df_to_join = df_to_join.sort_values( |
| 1200 | by=created_timestamp_column, na_position="first" |
| 1201 | ) |
| 1202 | df_to_join = df_to_join.persist() |
| 1203 | |
| 1204 | df_to_join = df_to_join.sort_values(by=timestamp_field, na_position="first") |
| 1205 | df_to_join = df_to_join.persist() |
| 1206 | |
| 1207 | except ZeroDivisionError: |
| 1208 | # Use 1 partition to get around case where everything in timestamp column is the same so the partition algorithm doesn't |
| 1209 | # try to divide by zero. |
| 1210 | if created_timestamp_column: |
| 1211 | df_to_join = df_to_join[column_order].sort_values( |
| 1212 | by=created_timestamp_column, na_position="first", npartitions=1 |
| 1213 | ) |
| 1214 | df_to_join = df_to_join.persist() |
| 1215 | |
| 1216 | df_to_join = df_to_join[column_order].sort_values( |
| 1217 | by=timestamp_field, na_position="first", npartitions=1 |
| 1218 | ) |
| 1219 | df_to_join = df_to_join.persist() |
| 1220 | |
| 1221 | df_to_join = df_to_join.drop_duplicates( |
| 1222 | all_join_keys + [entity_df_event_timestamp_col], |
| 1223 | keep="last", |
| 1224 | ignore_index=True, |
| 1225 | ) |
| 1226 | |
| 1227 | return df_to_join.persist() |
| 1228 | |
| 1229 | |
| 1230 | def _drop_columns( |
no test coverage detected