账户合并

from pyspark import SparkContext

# 初始化 SparkContext
sc = SparkContext("local", "AccountMerger")

# 读取输入数据
input_data = [
    "John johnsmith@mail.com john00@mail.com",
    "John johnnybravo@mail.com",
    "John johnsmith@mail.com john_newyork@mail.com",
    "Mary mary@mail.com"
]

# 将输入数据转换为 RDD
rdd = sc.parallelize(input_data)

# 解析数据，生成 (email, name) 的键值对
def parse_account(line):
    parts = line.split()
    name = parts[0]
    emails = parts[1:]
    return [(email, name) for email in emails]

email_name_pairs = rdd.flatMap(parse_account)

# 使用 groupByKey 合并账户
name_emails = email_name_pairs.groupByKey().mapValues(list)

# 构建最终结果
def build_result(grouped_emails):
    email_to_names = {}
    for email, names in grouped_emails:
        for name in names:
            if name not in email_to_names:
                email_to_names[name] = set()
            email_to_names[name].add(email)
    
    result = []
    for name, emails in email_to_names.items():
        sorted_emails = sorted(emails)
        result.append((name, sorted_emails))
    
    return result

final_result = build_result(name_emails.collect())

# 按账户名称排序
final_result.sort(key=lambda x: x[0])

# 输出结果
for name, emails in final_result:
    print(name, *emails)

# 停止 SparkContext
sc.stop()