3000万行大数据按ID分组、按日期排序生成值列表的高效实现问询

Hey there! Let's tackle this large-scale data processing problem properly—30 million rows is no joke, so we need to skip the slow apply/for loops entirely. Here's how to get this done efficiently, plus framework recommendations for your scale:

First: Efficient Pandas Implementation (if data fits in memory) #

Pandas' groupby can handle this way faster than apply if we structure the steps correctly, since we'll leverage vectorized operations instead of per-group Python function calls:

1. Fix the date column first: Make sure your date is a proper datetime type (string sorting might work for your example format, but it's risky for edge cases):

   import pandas as pd
   df = pd.read_csv('your_data.csv')
   # Adjust the format to match your actual date string (dd/mm/yyyy here)
   df['date'] = pd.to_datetime(df['date'], format='%d/%m/%Y')
   

2. Sort the entire dataframe first: By sorting on id and date upfront, we ensure that when we group, each group's values are already in the correct order:

   df_sorted = df.sort_values(['id', 'date'])
   

3. Group and aggregate with list: The agg(list) operation is optimized in pandas (runs in C-level code, not Python loops), so it's way faster than apply:

   result = df_sorted.groupby('id')['value'].agg(list).reset_index(name='VALUE_LIST_ORDERED')
   

This should handle 30M rows easily if you have enough RAM (rough estimate: each row with 3 columns is ~50 bytes, so 30M rows is ~1.5GB—most modern machines can handle that).

If Pandas Isn't Enough (data exceeds memory) #

For datasets too big for a single machine's RAM, use these frameworks:

Option 1: Dask (Pandas-like, out-of-core processing)

Dask mimics pandas syntax but processes data in chunks, so it works with data larger than memory:

import dask.dataframe as dd

# Read data with chunking, parse dates on load
ddf = dd.read_csv(
    'your_data.csv',
    parse_dates=['date'],
    date_format='%d/%m/%Y'
)

# Sort, group, aggregate—same logic as pandas
ddf_sorted = ddf.sort_values(['id', 'date'])
result_ddf = ddf_sorted.groupby('id')['value'].agg(list).reset_index(name='VALUE_LIST_ORDERED')

# Compute to get a pandas dataframe (or write directly to file)
result = result_ddf.compute()

Option 2: PySpark (Distributed processing for massive datasets)

If you're dealing with truly massive data (TB scale) or need cluster processing, PySpark is the way to go. It distributes the work across multiple nodes:

from pyspark.sql import SparkSession
from pyspark.sql.functions import collect_list, to_date

# Initialize Spark session
spark = SparkSession.builder.appName('id_value_list').getOrCreate()

# Load data and convert date column
df_spark = spark.read.csv('your_data.csv', header=True, inferSchema=True)
df_spark = df_spark.withColumn('date', to_date('date', 'dd/MM/yyyy'))

# Order by id and date first, then group and collect values
result_spark = (
    df_spark
    .orderBy('id', 'date')
    .groupBy('id')
    .agg(collect_list('value').alias('VALUE_LIST_ORDERED'))
)

# Show results or write to output
result_spark.show()
# result_spark.write.csv('output.csv', header=True)

Key Notes #

• Always use datetime types for dates: String sorting can fail if your date format changes (e.g., leading zeros missing), so converting to datetime avoids that risk.
• Avoid apply at all costs: For large datasets, apply is orders of magnitude slower because it runs Python code per group—stick to built-in aggregation functions like agg(list).

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