如何基于每日重置的tx_id等字段生成全局唯一索引列？

核心需求是生成同一tx_date下的同一tx_id对应所有行共用的全局递增id，以下分别提供Pandas和PySpark的实现方案：

Pandas 实现 #

利用组合键编码的方式，为每个唯一的(tx_date, tx_id)组合分配连续递增的id。

代码示例 #

import pandas as pd

# 示例数据（替换为你的实际数据即可）
data = {
    'tx_id': [1,1,2,3,4,4,1,1,2,2,3],
    'tx_date': ['1/4/2023']*6 + ['1/5/2023']*5,
    'product_id': [3,5,3,1,1,5,5,1,4,6,1]
}
df = pd.DataFrame(data)

# 按日期和事务id排序，保证id顺序符合预期
df_sorted = df.sort_values(['tx_date', 'tx_id'])
# 对日期+事务id的组合键生成连续编码，从1开始计数
df_sorted['id'] = pd.factorize(df_sorted[['tx_date', 'tx_id']].apply(tuple, axis=1))[0] + 1

# 查看结果
print(df_sorted)

关键说明 #

1. 先排序确保id的递增顺序与日期、事务id的逻辑一致；
2. 将tx_date和tx_id合并为元组作为唯一标识，通过factorize()生成从0开始的连续编码，加1后得到从1起始的id；
3. 同一(tx_date, tx_id)组合的所有行将共享同一个id。

PySpark 实现 #

使用窗口函数dense_rank()，直接对全局数据按tx_date和tx_id排序后生成连续排名，作为新id。

代码示例 #

from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import dense_rank

# 初始化Spark会话
spark = SparkSession.builder.appName("GlobalTxID").getOrCreate()

# 示例数据（替换为你的实际数据）
data = [
    (1, '1/4/2023', 3), (1, '1/4/2023', 5), (2, '1/4/2023', 3),
    (3, '1/4/2023', 1), (4, '1/4/2023', 1), (4, '1/4/2023', 5),
    (1, '1/5/2023', 5), (1, '1/5/2023', 1), (2, '1/5/2023', 4),
    (2, '1/5/2023', 6), (3, '1/5/2023', 1)
]
df = spark.createDataFrame(data, schema=["tx_id", "tx_date", "product_id"])

# 定义全局窗口：按日期和事务id排序
window_spec = Window.orderBy("tx_date", "tx_id")

# 生成id列
df_with_id = df.withColumn("id", dense_rank().over(window_spec))

# 展示结果
df_with_id.show()

关键说明 #

1. dense_rank()会对相同的(tx_date, tx_id)组合分配相同的排名，且排名连续无跳号；
2. 无需额外分组操作，直接通过全局窗口排序生成符合要求的id；
3. 最终id会严格按照日期先后、同日期内事务id递增的顺序生成。

内容的提问来源于stack exchange，提问作者Bagas Kara