如何在Dask/DuckDB中按条件格式化Parquet数据的Result字段？

1. Dask高效实现方式 #

你之前的代码遗漏行是因为链式索引（df[df["Count"] != 500]["Result"]）在Dask的惰性计算模型中无法正确修改原DataFrame，直接对列进行条件赋值即可解决问题。

实现代码： #

import dask.dataframe as dd

# 读取Parquet数据
df = dd.read_parquet("your_input_data.parquet")

# 构造500个32767拼接的字符串（无需依赖numpy，更简洁）
fill_str = ",".join(["32767"] * 500)

# 条件替换：仅保留Count=500的原Result，其余替换为目标字符串
df["Result"] = df["Result"].where(df["Count"] == 500, fill_str)

# 保存处理后的数据
df.to_parquet("processed_data.parquet", write_index=False)

优势： #

• 基于Dask惰性计算，无需加载全量数据到内存，适配大文件场景
• 条件替换逻辑直接覆盖所有符合要求的行，避免遗漏问题

2. 用SQL工具（DuckDB/SQLite）实现 #

单机场景下，SQL工具（尤其是DuckDB）处理这类批量替换效率很高，语法直观易维护。

方案一：DuckDB（推荐，直接支持Parquet） #

DuckDB可直接读取Parquet文件，无需提前导入，执行SQL后直接导出结果：

-- 创建处理后的表
CREATE OR REPLACE TABLE processed_data AS
SELECT
    Name,
    Count,
    CASE
        WHEN Count = 499 THEN (SELECT STRING_AGG('32767', ',') FROM range(500))
        ELSE Result
    END AS Result
FROM read_parquet('your_input_data.parquet');

-- 导出为Parquet文件
COPY processed_data TO 'processed_data.parquet' (FORMAT PARQUET);

方案二：SQLite #

SQLite本身不支持直接读取Parquet，需先通过工具中转导入数据，再执行替换：

步骤1：将Parquet导入SQLite（用pandas中转）

import pandas as pd
import sqlite3

# 读取Parquet数据（文件过大时建议用Dask分批写入）
df = pd.read_parquet("your_input_data.parquet")

# 连接SQLite数据库
conn = sqlite3.connect("data.db")

# 将数据写入SQLite表
df.to_sql("raw_data", conn, index=False, if_exists="replace")

步骤2：执行SQL替换

-- 生成处理后的表
CREATE TABLE processed_data AS
SELECT
    Name,
    Count,
    CASE
        WHEN Count = 499 THEN (
            SELECT GROUP_CONCAT('32767', ',')
            FROM (
                WITH nums AS (
                    SELECT 1 AS n
                    UNION ALL
                    SELECT n+1 FROM nums WHERE n < 500
                )
                SELECT * FROM nums
            )
        )
        ELSE Result
    END AS Result
FROM raw_data;

步骤3：导出为Parquet

# 读取处理后的数据
processed_df = pd.read_sql("SELECT * FROM processed_data", conn)

# 保存为Parquet文件
processed_df.to_parquet("processed_data.parquet", index=False)

# 关闭数据库连接
conn.close()

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