在R中分块处理大型gzip压缩JSON Lines文件并导入DuckDB

核心思路 #

• 用Python标准库gzip逐行读取压缩文件，避免一次性加载全量数据占满内存
• 扁平化OpenAlex作者JSON的嵌套结构，拆分为主表+两个关联子表
• 按固定行数分块写入CSV，严格控制内存占用
• 最后用DuckDB批量导入生成的CSV文件

完整代码示例 #

import gzip
import json
import csv
import duckdb

# 配置参数
INPUT_GZIP = "part_000.gz"  # 替换为你的本地文件路径
CHUNK_SIZE = 10000  # 每块处理的行数，可根据内存调整
# 输出CSV路径
AUTHORS_CSV = "authors.csv"
AUTHOR_IDS_CSV = "author_ids.csv"
AUTHOR_COUNTS_CSV = "author_counts.csv"

# 扁平化单个作者JSON对象，拆分出三个表的数据
def flatten_author(author_json):
    # 提取OpenAlex ID作为关联主键
    author_id = author_json["id"].split("/")[-1]
    
    # 主表：作者基础信息
    author_main = {
        "author_id": author_id,
        "display_name": author_json.get("display_name"),
        "orcid": author_json.get("orcid"),
        "updated_date": author_json.get("updated_date"),
        "created_date": author_json.get("created_date")
    }
    
    # 子表：作者的各类ID（openalex、orcid等）
    author_ids = []
    for id_type, id_value in author_json.get("ids", {}).items():
        author_ids.append({
            "author_id": author_id,
            "id_type": id_type,
            "id_value": id_value
        })
    
    # 子表：作者的统计数据
    author_counts = []
    counts = author_json.get("counts", {})
    author_counts.append({"author_id": author_id, "count_type": "works_count", "count_value": counts.get("works_count")})
    author_counts.append({"author_id": author_id, "count_type": "cited_by_count", "count_value": counts.get("cited_by_count")})
    
    return author_main, author_ids, author_counts

# 初始化CSV文件，写入表头
def init_csvs():
    with open(AUTHORS_CSV, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=["author_id", "display_name", "orcid", "updated_date", "created_date"])
        writer.writeheader()
    
    with open(AUTHOR_IDS_CSV, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=["author_id", "id_type", "id_value"])
        writer.writeheader()
    
    with open(AUTHOR_COUNTS_CSV, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=["author_id", "count_type", "count_value"])
        writer.writeheader()

# 分块读取并处理压缩文件，写入CSV
def process_file():
    init_csvs()
    chunk_main = []
    chunk_ids = []
    chunk_counts = []
    
    with gzip.open(INPUT_GZIP, "rt", encoding="utf-8") as f:
        for line_num, line in enumerate(f, 1):
            if not line.strip():
                continue
            try:
                author = json.loads(line)
                main, ids, counts = flatten_author(author)
                chunk_main.append(main)
                chunk_ids.extend(ids)
                chunk_counts.extend(counts)
                
                # 达到块大小则写入并清空缓存
                if line_num % CHUNK_SIZE == 0:
                    # 写入主表
                    with open(AUTHORS_CSV, "a", newline="", encoding="utf-8") as f_main:
                        writer = csv.DictWriter(f_main, fieldnames=chunk_main[0].keys())
                        writer.writerows(chunk_main)
                    # 写入ID子表
                    with open(AUTHOR_IDS_CSV, "a", newline="", encoding="utf-8") as f_ids:
                        writer = csv.DictWriter(f_ids, fieldnames=chunk_ids[0].keys())
                        writer.writerows(chunk_ids)
                    # 写入统计子表
                    with open(AUTHOR_COUNTS_CSV, "a", newline="", encoding="utf-8") as f_counts:
                        writer = csv.DictWriter(f_counts, fieldnames=chunk_counts[0].keys())
                        writer.writerows(chunk_counts)
                    
                    chunk_main = []
                    chunk_ids = []
                    chunk_counts = []
                    print(f"已处理 {line_num} 行")
            except Exception as e:
                print(f"处理第 {line_num} 行出错: {e}")
    
    # 处理剩余的不足一块的数据
    if chunk_main:
        with open(AUTHORS_CSV, "a", newline="", encoding="utf-8") as f_main:
            writer = csv.DictWriter(f_main, fieldnames=chunk_main[0].keys())
            writer.writerows(chunk_main)
    if chunk_ids:
        with open(AUTHOR_IDS_CSV, "a", newline="", encoding="utf-8") as f_ids:
            writer = csv.DictWriter(f_ids, fieldnames=chunk_ids[0].keys())
            writer.writerows(chunk_ids)
    if chunk_counts:
        with open(AUTHOR_COUNTS_CSV, "a", newline="", encoding="utf-8") as f_counts:
            writer = csv.DictWriter(f_counts, fieldnames=chunk_counts[0].keys())
            writer.writerows(chunk_counts)
    print("文件处理完成")

# 将CSV导入DuckDB
def import_to_duckdb():
    conn = duckdb.connect("openalex_authors.db")
    
    # 创建主表并导入
    conn.execute("""
        CREATE TABLE IF NOT EXISTS authors (
            author_id VARCHAR PRIMARY KEY,
            display_name VARCHAR,
            orcid VARCHAR,
            updated_date DATE,
            created_date DATE
        )
    """)
    conn.execute(f"COPY authors FROM '{AUTHORS_CSV}' (HEADER, DELIMITER ',', ENCODING 'UTF-8')")
    
    # 创建ID子表
    conn.execute("""
        CREATE TABLE IF NOT EXISTS author_ids (
            author_id VARCHAR,
            id_type VARCHAR,
            id_value VARCHAR,
            PRIMARY KEY (author_id, id_type),
            FOREIGN KEY (author_id) REFERENCES authors(author_id)
        )
    """)
    conn.execute(f"COPY author_ids FROM '{AUTHOR_IDS_CSV}' (HEADER, DELIMITER ',', ENCODING 'UTF-8')")
    
    # 创建统计子表
    conn.execute("""
        CREATE TABLE IF NOT EXISTS author_counts (
            author_id VARCHAR,
            count_type VARCHAR,
            count_value INTEGER,
            PRIMARY KEY (author_id, count_type),
            FOREIGN KEY (author_id) REFERENCES authors(author_id)
        )
    """)
    conn.execute(f"COPY author_counts FROM '{AUTHOR_COUNTS_CSV}' (HEADER, DELIMITER ',', ENCODING 'UTF-8')")
    
    # 验证导入结果
    print("导入后数据量：")
    print("authors表行数:", conn.execute("SELECT COUNT(*) FROM authors").fetchone()[0])
    print("author_ids表行数:", conn.execute("SELECT COUNT(*) FROM author_ids").fetchone()[0])
    print("author_counts表行数:", conn.execute("SELECT COUNT(*) FROM author_counts").fetchone()[0])
    
    conn.close()

if __name__ == "__main__":
    process_file()
    import_to_duckdb()

使用说明 #

1. 先将示例文件下载到本地，修改INPUT_GZIP为本地文件路径
2. 调整CHUNK_SIZE：内存充足可调大，内存紧张则调小（比如5000）
3. 安装依赖：执行pip install duckdb（gzip、json、csv为Python标准库，无需额外安装）
4. 运行脚本：自动生成三个CSV文件，随后导入到DuckDB数据库

关键细节 #

• 用gzip.open以文本模式读取压缩文件，逐行解析JSON，完全避免加载全量数据
• 提取OpenAlex ID作为关联主键，保证三个表的数据一致性
• 分块写入CSV时用追加模式，不会覆盖已写入内容
• DuckDB导入时自动识别表头，支持UTF-8编码处理特殊字符

内容的提问来源于stack exchange，提问作者Karsten W.