ranking_column是什么时候赋值的

在deduplicate函数中，ranking_column是一个参数，它指定了用于标识重复记录的列的名称。具体来说，该参数是在调用add_duplicates_rank函数时传入的，如下所示：

df = add_duplicates_rank(df, key_columns, order_by_columns, ranking_column)

其中，ranking_column是一个字符串，表示用于标识重复记录的列的名称，它可以是任何合法的列名。在add_duplicates_rank函数中，会使用pyspark的Window函数对DataFrame进行分组、排序、排名等操作，并将生成的重复排名值作为新的一列添加到DataFrame中，列名就是传入的ranking_column参数。最终，add_duplicates_rank函数返回一个新的DataFrame，其中包含了新添加的"duplicates_rank"列。

相关问题

ranking_column = "duplicates_rank"为什么这么赋值，并且进行与1比较

在deduplicate函数中，ranking_column参数默认值为"duplicates_rank"，这是因为在进行数据去重时，我们需要为每个分组内的记录生成一个排名值，用于标识重复记录的排名。由于这个排名值是在函数内部生成的，因此我们需要为它指定一个列名。为了方便使用，我们默认将这个列名命名为"duplicates_rank"。

在去重完成后，我们需要将"duplicates_rank"列中排名为1的记录保留下来，而将其他重复记录删除。因此，我们需要将"duplicates_rank"列中的每个值与1进行比较，以确定哪些记录是重复记录，哪些是唯一记录。如果"duplicates_rank"列中的值为1，则表示该记录是唯一记录，否则就是重复记录。

当然，如果需要使用其他名称来标识重复记录的排名，也可以在调用deduplicate函数时通过传入ranking_column参数来指定。而将"duplicates_rank"列中的值与1进行比较，是因为在deduplicate函数中，我们默认将重复排名值按升序排序，因此排名为1的记录就是每个分组内的第一个记录。

if (self.config.filter_deleted_flag) & ( is_deleted_column in df.columns): # if True removes records that marked deleted filter = (F.col(ranking_column) == 1) & (F.col(is_deleted_column) == "False") else: self.logger.warning(f"Records marked as deleted will be loaded to {self.config.table_full_name} table!") filter = F.col(ranking_column) == 1 return df.withColumn(ranking_column, F.row_number().over(Window.partitionBy(key_columns).orderBy(order_by_columns))).filter( filter).drop(ranking_column)

这段代码是一个Python函数，主要功能是对DataFrame进行过滤、排序、计算排名，并返回结果。具体来说，函数根据一些参数（如是否过滤已删除的记录、排序方式等）对DataFrame进行过滤和排序，然后计算每个分组中每行的排名，最后返回过滤后的结果。具体注释如下：

def filter_sort_rank(df, key_columns, order_by_columns, ranking_column, is_deleted_column='is_deleted'):
    # 如果需要过滤已删除的记录，则将is_deleted_column列为"False"的记录过滤掉
    if (self.config.filter_deleted_flag) & (is_deleted_column in df.columns):
        filter = (F.col(ranking_column) == 1) & (F.col(is_deleted_column) == "False")
    else:
        # 如果不需要过滤已删除的记录，则保留所有记录，并发出警告
        self.logger.warning(f"Records marked as deleted will be loaded to {self.config.table_full_name} table!")
        filter = F.col(ranking_column) == 1
    # 对DataFrame进行排序、计算排名，并过滤掉不符合条件的记录
    return df.withColumn(ranking_column, F.row_number().over(Window.partitionBy(key_columns).orderBy(order_by_columns))).filter(filter).drop(ranking_column)

其中，`df`是输入的DataFrame，`key_columns`指定了分组的列，`order_by_columns`指定了排序的列，`ranking_column`指定了计算排名时添加的新列名，`is_deleted_column`指定了标记是否删除的列名。