从HDFS多目录Parquet文件创建Dask DataFrame的最优方案咨询

Hey there! Let's figure out how to speed up creating that Dask DataFrame from all those HDFS Parquet files—those slow read times are no fun.

First, let's break down why your current methods might be dragging their feet, then dive into actionable fixes:

Why Your Current Methods Are Slow #

• Recursive Glob Overhead: Using ** in your path makes Dask traverse every subdirectory on HDFS, which means tons of metadata requests to the namenode. With thousands of files/dirs, this gets painfully slow.
• Concat Overhead: Creating separate DataFrames per directory and concatenating them adds extra work for schema alignment and task graph merging—especially with many small DataFrames, this adds unnecessary latency.

Actionable Speedups #

1. Pre-Generate File Paths (Skip Recursive Glob) #

Instead of letting Dask crawl HDFS, get the full list of Parquet files yourself first using an HDFS client, then pass that list directly to read_parquet. This cuts out most of the metadata lookup delay.

Example with hdfs3:

from hdfs3 import HDFileSystem
import dask.dataframe as dd

# Set up HDFS connection
hdfs = HDFileSystem(host="your-namenode-host", port=9000)

# Get all parquet files (adjust the glob pattern if you can avoid full recursion!)
file_paths = [f"hdfs://{hdfs.host}:{hdfs.port}/{path}" for path in hdfs.glob("/your/root/path/**/*.parquet")]

# Read directly with the precomputed list
df = dd.read_parquet(file_paths, engine="pyarrow")

2. Reuse HDFS Connections #

Using persistent HDFS connections (instead of creating new ones for each read) reduces overhead. PyArrow's HDFS integration handles this nicely:

import pyarrow.hdfs as phdfs
import dask.dataframe as dd

# Create a single persistent HDFS connection
hdfs_conn = phdfs.connect(host="your-namenode-host", port=9000)

# Use this connection for reading
df = dd.read_parquet(
    "/your/root/path/**/*.parquet",
    engine="pyarrow",
    storage_options={"hdfs": hdfs_conn},
    recursive=True
)

3. Fix "Small File Hell" #

If you have thousands of tiny Parquet files, Dask creates a task per file, which overwhelms the scheduler. Fix this by:

• Coalescing files first: Use dask.delayed to batch small files into larger ones before reading.
• Avoiding unnecessary recursion: If you know your directory structure, use explicit glob patterns (like /path/to/dir-*/files*.parquet) instead of ** to skip irrelevant directories.

4. Parallelize Metadata Collection #

Instead of collecting file paths sequentially, use Dask's delayed to parallelize the work:

from dask import delayed, compute
泰戈尔,弹 SCCalculatePl汉初Orange一杆 highly前后 devise拿出自己的？不对，重新写：
from dask import delayed, compute
import dask.dataframe as dd

@delayed
def get_files(dir_path):
    hdfs = HDFileSystem(host="your-namenode-host", port=9000)
    return hdfs.glob(f"{dir_path}/*.parquet")

# List your top-level directories
top_dirs = ["/path/to/dir1", "/path/to/dir2", "/path/to/dir3"]
# Compute all file paths in parallel
file_lists = compute(*[get_files(d) for d in top_dirs])
# Flatten into a single list
all_files = [file for sublist in file_lists for file in sublist]

df = dd.read_parquet(all_files)

5. Optimize Schema & Engine #

• Use PyArrow instead of FastParquet: PyArrow's HDFS integration is generally faster for large-scale reads.
• Specify schema explicitly: If you know your data's schema upfront, pass it to read_parquet to skip schema inference (which requires reading a sample from each file):

import pyarrow as pa

# Define your schema
schema = pa.schema([
    ("user_id", pa.int64()),
    ("timestamp", pa.timestamp("ns")),
    ("value", pa.float64())
])

df = dd.read_parquet(all_files, schema=schema, engine="pyarrow")

Bonus HDFS Performance Tips #

• Prune columns: Only read the columns you need with the columns parameter in read_parquet—this cuts down on data transfer.
• Tune Dask workers: If you're hitting resource bottlenecks, increase worker CPU/memory to handle read tasks more efficiently.
• Use HDFS Federation: If your cluster uses federation, connect to the correct namenode for each directory to avoid cross-namenode metadata hops.

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