Dask DataFrame入門

Daskとは

Daskのインストールとインポート

import dask.dataframe as dd

# Read a large CSV — returns a Dask DataFrame immediately (no data loaded yet)
ddf = dd.read_csv('large_sales.csv')
print(type(ddf))     # dask.dataframe.core.DataFrame
print(ddf.columns.tolist())
print(ddf.dtypes)

DaskとPandasの違い：重要なポイント

import dask.dataframe as dd

ddf = dd.read_csv('sales.csv')

# This does NOT run yet — just builds the task graph
filtered = ddf[ddf['amount'] > 1000]
agg = filtered.groupby('region')['amount'].sum()

print(type(agg))  # dask.dataframe.core.Series

# NOW execute everything
result = agg.compute()
print(result)

パーティション：中心となる概念

import dask.dataframe as dd

ddf = dd.read_csv('data/*.csv')  # Read multiple CSV files at once
print('Number of partitions:', ddf.npartitions)

# Access a single partition as a Pandas DataFrame
first_partition = ddf.get_partition(0).compute()
print('Partition 0 shape:', first_partition.shape)

Daskで使えるおなじみのPandas操作

import dask.dataframe as dd

ddf = dd.read_csv('transactions.csv')

# Filtering — same syntax as Pandas
high_value = ddf[ddf['amount'] > 500]

# GroupBy aggregation
by_region = high_value.groupby('region')['amount'].mean()

# Execute
result = by_region.compute()
print(result.sort_values(ascending=False))

Globパターンで複数ファイルを読み込む

import dask.dataframe as dd

# Read all monthly files at once
ddf = dd.read_csv('sales/2024-*.csv',
                  dtype={'order_id': 'int32',
                         'amount': 'float32'})
print(f'Partitions: {ddf.npartitions}')  # One per file
print(f'Total rows (lazy): {len(ddf)}')  # This triggers a compute!

タスクグラフのvisualize()メソッド

import dask.dataframe as dd

ddf = dd.read_csv('orders.csv')
pipeline = (
    ddf[ddf['status'] == 'completed']
    .groupby('product_id')['revenue']
    .sum()
)

# Visualise the task graph (saves to PNG)
# pipeline.visualize('task_graph.png')

# Check number of tasks in the graph
print('Number of tasks:', len(pipeline.__dask_graph__()))

map_partitionsでカスタム関数を適用する

import dask.dataframe as dd
import pandas as pd

def normalise_chunk(df):
    df = df.copy()
    df['amount_norm'] = (df['amount'] - df['amount'].mean()) / df['amount'].std()
    return df

ddf = dd.read_csv('data.csv')
normalised = ddf.map_partitions(normalise_chunk)
result = normalised[['order_id', 'amount_norm']].compute()
print(result.head())

Daskのスケジューラーオプション

import dask.dataframe as dd

ddf = dd.read_csv('data.csv')
agg = ddf.groupby('category')['sales'].sum()

# Choose scheduler based on workload
result_sync = agg.compute(scheduler='synchronous')  # sequential, easy to debug
result_threads = agg.compute(scheduler='threads')   # parallel I/O
result_processes = agg.compute(scheduler='processes')  # parallel CPU

DaskとPandasの間で変換する

import pandas as pd
import dask.dataframe as dd

# Start with a small Pandas DF for testing
df_small = pd.DataFrame({'a': range(100), 'b': range(100, 200)})

# Convert to Dask for development/testing
ddf = dd.from_pandas(df_small, npartitions=4)
result = ddf.groupby('a')['b'].sum().compute()
print(type(result))  # pandas.Series
print(result.head())

DaskとPandasとSQLの使い分け

理解度チェック

レッスンのまとめ