350万用户与14万交易数据匹配的性能优化方案求助

问题背景 #

我拥有两个数组：包含350万条用户数据的users数组，以及包含14万条交易数据的transactions数组。需要遍历每个用户，统计其完成的交易类型（membership、credits），结果存入包含350万条记录的新数组Array C，后续用该数组生成点云。

当前采用嵌套循环实现（代码如下），但耗时约2小时，希望找到更高效的实现方案或工具库（已了解numpy）：

with open('final/users.json', 'r') as f:
  users = json.load(f)

with open('final/transactions.json', 'r') as f:
  transactions = json.load(f)

result = []
count = 0

for user in users:
  data_elem = {'user': user, 'memberships': 0, 'credits': 0}
  print('working on: ' + str(user['id']) + ' count: ' + str(count))
  for transaction in transactions:
    if user['id'] == transaction['user_id']:
      if transaction['type'] == 'membership':
        data_elem['memberships'] += 1
      elif transaction['type'] == 'credits':
        data_elem['credits'] += 1
      else:
        print('unknow type: ' + transaction['type'])
        break
  result.append(data_elem)
  count += 1
  # Just for testing
  if count == 2000:
    break

with open('final/analysed/membership_credits_buys.json', 'w') as f:
  json.dump(result, f)

优化方案 #

方案1：字典预统计交易数据（纯Python实现，无需额外库） #

嵌套循环的时间复杂度是O(M*N)（M为用户数，N为交易数），这是效率低下的核心原因。可以先遍历一次交易数据，用字典预统计每个用户的交易类型计数，再遍历用户数组匹配结果，时间复杂度降到O(M+N)，效率会大幅提升：

import json

# 加载数据
with open('final/users.json', 'r') as f:
  users = json.load(f)

with open('final/transactions.json', 'r') as f:
  transactions = json.load(f)

# 预统计每个用户的交易计数
transaction_counts = {}
for trans in transactions:
    user_id = trans['user_id']
    trans_type = trans['type']
    if user_id not in transaction_counts:
        transaction_counts[user_id] = {'memberships': 0, 'credits': 0}
    if trans_type == 'membership':
        transaction_counts[user_id]['memberships'] += 1
    elif trans_type == 'credits':
        transaction_counts[user_id]['credits'] += 1
    # 未知类型可选择记录或忽略，这里跳过打印避免影响效率

# 生成结果数组
result = []
for idx, user in enumerate(users):
    user_id = user['id']
    counts = transaction_counts.get(user_id, {'memberships': 0, 'credits': 0})
    data_elem = {
        'user': user,
        'memberships': counts['memberships'],
        'credits': counts['credits']
    }
    result.append(data_elem)
    # 可选：每处理一定数量用户打印进度，避免频繁打印拖慢速度
    if idx % 100000 == 0:
        print(f"Processed {idx} users")

# 保存结果
with open('final/analysed/membership_credits_buys.json', 'w') as f:
  json.dump(result, f)

方案2：使用Pandas处理（适合大数据集+后续点云生成） #

Pandas对结构化数据的分组、合并操作做了高度优化，效率远高于纯Python循环，而且后续生成点云时可以直接用DataFrame的数据，无需额外转换：

import json
import pandas as pd

# 加载数据为DataFrame
with open('final/users.json', 'r') as f:
    users_df = pd.DataFrame(json.load(f))

with open('final/transactions.json', 'r') as f:
    trans_df = pd.DataFrame(json.load(f))

# 统计每个用户的交易类型数量
# 先筛选有效交易类型，再分组计数
valid_trans = trans_df[trans_df['type'].isin(['membership', 'credits'])]
counts_df = valid_trans.groupby(['user_id', 'type']).size().unstack(fill_value=0)
counts_df = counts_df.rename(columns={'membership': 'memberships', 'credits': 'credits'}).reset_index()

# 和用户数据合并，确保所有用户都有记录（无交易的用户计数为0）
result_df = users_df.merge(counts_df, left_on='id', right_on='user_id', how='left').fillna(0)
# 转换为需要的结构（如果必须和原格式一致）
result = result_df.apply(
    lambda row: {
        'user': row.drop(['user_id', 'memberships', 'credits']).to_dict(),
        'memberships': int(row['memberships']),
        'credits': int(row['credits'])
    },
    axis=1
).tolist()

# 保存结果
with open('final/analysed/membership_credits_buys.json', 'w') as f:
    json.dump(result, f)

# 后续生成点云可直接用result_df的memberships和credits列
# 例如：points = result_df[['memberships', 'credits']].values

额外优化：合并点云生成步骤 #

如果后续生成点云只需要用户的memberships和credits计数，可以在第一次统计时直接收集这些数据，不用先保存完整的Array C，节省内存和时间：

比如方案1中，预统计后可以直接生成点云数据：

# 生成点云数据（假设点云需要的是[memberships, credits]的二维数组）
point_cloud = []
for user in users:
    counts = transaction_counts.get(user['id'], {'memberships': 0, 'credits': 0})
    point_cloud.append([counts['memberships'], counts['credits']])

# 后续直接使用point_cloud即可，无需先保存result数组

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