如何对分别存储问题和对应多条答案的两个文件执行内连接操作

方案1：小文件场景直接用awk命令实现 #

一行命令即可得到目标结果，无需写复杂逻辑：

awk -F ',' 'NR==FNR{q[$1]=$0;next} {print q[$2]" "$1","$3}' 问题文件.csv 答案文件.csv

逻辑说明：

• 优先读取第一个文件（问题文件），以QID为key，整行内容存入字典q
• 再读取第二个文件（答案文件），按第二列的QID匹配对应问题内容，按要求拼接输出即可

方案2：MapReduce实现逻辑（适合大数据量场景） #

你之前的思路方向正确，调整细节后即可跑通，完整步骤如下：

Map阶段 #

读取文件时先区分当前行是问题还是答案：

• 问题行（行开头为Q+数字）：按逗号拆分后，输出key为QID（如Q1），value加前缀标记q#+完整问题行内容，例：q#Q1,What is the name of your son?
• 答案行（行开头为A+数字）：按逗号拆分后，输出key为第二列的QID（如Q1），value加前缀标记a#+AID+答案内容，例：a#A1,George

Reduce阶段 #

同一个QID的所有value会被分到同一个Reduce任务处理：

• 先遍历所有value，提取出带q#前缀的问题内容单独存储
• 再遍历所有带a#前缀的答案内容，将问题内容与答案内容按要求拼接输出即可

Python版示例代码 #

Map端代码：

import sys
for line in sys.stdin:
    line = line.strip()
    parts = line.split(',')
    if parts[0].startswith('Q'):
        qid = parts[0]
        print(f"{qid}\tq#{line}")
    else:
        qid = parts[1]
        a_info = f"{parts[0]},{parts[2]}"
        print(f"{qid}\ta#{a_info}")

Reduce端代码：

import sys
current_qid = None
q_content = ""
a_list = []
for line in sys.stdin:
    line = line.strip()
    qid, value = line.split('\t', 1)
    if current_qid != qid:
        if current_qid and q_content:
            for a in a_list:
                print(f"{q_content} {a}")
        current_qid = qid
        q_content = ""
        a_list = []
    if value.startswith('q#'):
        q_content = value[2:]
    elif value.startswith('a#'):
        a_list.append(value[2:])
# 处理最后一组QID的输出
if current_qid and q_content:
    for a in a_list:
        print(f"{q_content} {a}")

本地测试命令：

cat 问题文件.csv 答案文件.csv | python map.py | sort | python reduce.py

大数据量场景下直接用Hadoop Streaming提交上述两个脚本运行即可。

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