def _mark_duplicates(self, subset: Optional[Union[(Name, List[Name])]]=None, keep: Union[(bool, str)]='first') -> Tuple[(SparkDataFrame, str)]:
    if (subset is None):
        subset_list = self._internal.column_labels
    else:
        if is_name_like_tuple(subset):
            subset_list = [cast(Label, subset)]
        elif is_name_like_value(subset):
            subset_list = [(subset,)]
        else:
            subset_list = [(sub if is_name_like_tuple(sub) else (sub,)) for sub in subset]
        diff = set(subset_list).difference(set(self._internal.column_labels))
        if (len(diff) > 0):
            raise KeyError(', '.join([name_like_string(d) for d in diff]))
    group_cols = [self._internal.spark_column_name_for(label) for label in subset_list]
    sdf = self._internal.resolved_copy.spark_frame
    column = verify_temp_column_name(sdf, '__duplicated__')
    if ((keep == 'first') or (keep == 'last')):
        if (keep == 'first'):
            ord_func = F.asc
        else:
            ord_func = F.desc
        window = Window.partitionBy(*group_cols).orderBy(ord_func(NATURAL_ORDER_COLUMN_NAME)).rowsBetween(Window.unboundedPreceding, Window.currentRow)
        sdf = sdf.withColumn(column, (F.row_number().over(window) > 1))
    elif (not keep):
        window = Window.partitionBy(*group_cols).rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)
        sdf = sdf.withColumn(column, (F.count('*').over(window) > 1))
    else:
        raise ValueError("'keep' only supports 'first', 'last' and False")
    return (sdf, column)
def _mark_duplicates(self, subset: Optional[Union[(Name, List[Name])]]=None, keep: Union[(bool, str)]='first') -> Tuple[(SparkDataFrame, str)]:
    if (subset is None):
        subset_list = self._internal.column_labels
    else:
        if is_name_like_tuple(subset):
            subset_list = [cast(Label, subset)]
        elif is_name_like_value(subset):
            subset_list = [(subset,)]
        else:
            subset_list = [(sub if is_name_like_tuple(sub) else (sub,)) for sub in subset]
        diff = set(subset_list).difference(set(self._internal.column_labels))
        if (len(diff) > 0):
            raise KeyError(', '.join([name_like_string(d) for d in diff]))
    group_cols = [self._internal.spark_column_name_for(label) for label in subset_list]
    sdf = self._internal.resolved_copy.spark_frame
    column = verify_temp_column_name(sdf, '__duplicated__')
    if ((keep == 'first') or (keep == 'last')):
        if (keep == 'first'):
            ord_func = F.asc
        else:
            ord_func = F.desc
        window = Window.partitionBy(*group_cols).orderBy(ord_func(NATURAL_ORDER_COLUMN_NAME)).rowsBetween(Window.unboundedPreceding, Window.currentRow)
        sdf = sdf.withColumn(column, (F.row_number().over(window) > 1))
    elif (not keep):
        window = Window.partitionBy(*group_cols).rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)
        sdf = sdf.withColumn(column, (F.count('*').over(window) > 1))
    else:
        raise ValueError("'keep' only supports 'first', 'last' and False")
    return (sdf, column)