如何用Python无需解压解析Wikidata的JSON(.bz2)转储文件？

当然可以！不用解压整个18GB的bz2文件，Python完全能帮你流式读取并解析Wikidata的JSON转储，只处理你需要的部分，不会占用过多内存。Wikidata的转储格式是每行一个JSON对象（包裹在一个大数组里），这刚好适合逐行流式处理。

基础方法：用bz2和json模块逐行解析 #

Python标准库的bz2模块支持直接打开并读取bz2压缩文件，不需要提前解压。结合json模块，我们可以逐行解析每个实体：

import bz2
import json
from pprint import pprint

# 替换成你的Wikidata转储文件路径
dump_file = 'wikidata-20240506-all.json.bz2'

with bz2.open(dump_file, 'rt', encoding='utf-8') as f:
    for line_num, line in enumerate(f):
        # 跳过JSON数组的开头[和结尾]
        stripped_line = line.strip()
        if stripped_line in ('[', ']'):
            continue
        # 移除行尾的逗号（因为数组元素之间用逗号分隔）
        cleaned_line = stripped_line.rstrip(',')
        
        try:
            # 解析当前行的实体JSON
            item = json.loads(cleaned_line)
            
            # 示例1：查找特定ID的实体（比如Q42，Douglas Adams）
            if item['id'] == 'Q42':
                print(f"找到目标实体：{item['labels']['en']['value']}")
                print("核心属性：")
                # 查看实体的类型（P31）
                if 'P31' in item['claims']:
                    entity_type = item['claims']['P31'][0]['mainsnak']['datavalue']['value']['id']
                    print(f"  实体类型：{entity_type}")
                # 查看出生日期（P569）
                if 'P569' in item['claims']:
                    birth_date = item['claims']['P569'][0]['mainsnak']['datavalue']['value']['time']
                    print(f"  出生日期：{birth_date}")
                # 打印完整实体结构（可选）
                # pprint(item)
                break  # 找到后退出，避免处理整个文件
            
            # 示例2：过滤特定类型的实体（比如人类，P31=Q5）
            # if 'claims' in item and 'P31' in item['claims']:
            #     for claim in item['claims']['P31']:
            #         if claim['mainsnak']['datavalue']['value']['id'] == 'Q5':
            #             print(f"人类实体：{item['id']} - {item.get('labels', {}).get('en', {}).get('value', '无英文名')}")
            #             # 找到10个就退出
            #             if line_num >= 10:
            #                 exit()
                    
        except json.JSONDecodeError as e:
            print(f"第{line_num}行解析失败：{str(e)}")
            continue

代码说明： #

• bz2.open(..., 'rt')：以文本模式打开压缩文件，自动解压缩每行内容
• 跳过数组首尾的[和]，处理行尾的逗号，避免JSON解析错误
• 通过item['id']或claims中的属性（比如P31表示实体类型）过滤目标实体
• 找到目标后立即break，避免不必要的计算

进阶方法：用ijson流式解析复杂JSON #

如果需要更高效地处理大型JSON对象（比如实体的嵌套属性），可以使用第三方库ijson，它能流式解析JSON的特定部分，不需要把整个对象加载到内存：

首先安装ijson：

pip install ijson

然后使用以下代码：

import bz2
import ijson

dump_file = 'wikidata-20240506-all.json.bz2'

with bz2.open(dump_file, 'rb') as f:
    # 流式解析数组中的每个item（Wikidata转储的每个实体对应一个item）
    for item in ijson.items(f, 'item'):
        if item['id'] == 'Q42':
            print(f"找到实体：{item['labels']['en']['value']}")
            print("所有属性：")
            for prop_id, claims in item['claims'].items():
                for claim in claims:
                    # 提取属性值（根据属性类型不同，value结构可能有差异）
                    value = claim['mainsnak']['datavalue']['value']
                    print(f"  {prop_id}: {value}")
            break

优势： #

• 内存占用极低，适合处理超大文件
• 可以精准定位JSON中的特定字段，不需要解析整个实体

注意事项 #

• Wikidata转储的格式可能会随版本略有变化，处理前可以先查看几行确认结构
• 如果需要批量提取数据，建议添加进度提示（比如每处理10000行打印一次）
• 避免一次性处理整个文件，尽量通过过滤条件缩小处理范围

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