The other day, I did A simple performance comparison of Pandas, Dask, and Vaex in CSV, Parquet, and HDF5 in a single file.

However, since I often handle a lot of files with time series data etc. in my usual work, this time I will roughly compare the performance including the number of lines that the memory is quite tight when handling multiple files and normally. To go.

TL;DR

――In the time series assumed data I tried this time, there was no difference between Vaex + uncompressed HDF5 and Vaex + Snappy compressed Parquet as I expected. Considering the ease of handling and file size, Parquet with Snappy compression seems to be a good choice. -* In the previous comparison with a single file, HDF5 was faster than Parquet when the number of lines in one file was much larger, so it seems that only one file contains a large amount of data. HDF5 seems to shine in the case. --Pandas and Dask don't seem to be very fast with HDF5 when there are many files in chronological data (especially Dask is significantly slower). ――As time series data, when totaling around 100,000 lines per file as verified this time, it seems that there is no extreme difference between Vaex and Dask (although Vaex is about twice as fast). If anything, the disk access aspect of the environment I tried this time may also be a bottleneck. ――However, it is very comfortable and helpful that the calculation can be completed in about one and a half minutes even if it is less than 200 million lines.

Environment to use

We will proceed with the following Docker image settings. The host is using a Windows 10 laptop.

FROM jupyter/datascience-notebook:python-3.8.6
RUN pip install vaex==2.6.1
RUN pip install jupyter-contrib-nbextensions==0.5.1
RUN jupyter contrib nbextension install --user
RUN jupyter nbextension enable hinterland/hinterland \
    && jupyter nbextension enable toc2/main

The OS and library environment is as follows.

--Ubuntu is 20.04

vaex==2.6.1
pandas==1.1.4
dask==2.30.0
Jupyter notebook

Target format

In the comparison of the previous article, it was like uncompressed HDF5 or Snappy compressed Parquet, so this time I will deal with two formats, HDF5 and Snappy compressed Parquet, without including CSV.

Also, regarding the library, if you want to compare the number of lines that can fit in the memory, including Pandas, if the number of lines becomes quite large, proceed with only Dask and Vaex.

I will touch on each code below, but if you want to skip it and check the result, I would appreciate it if you could skip to this section.

Data preparation

Code from the previous performance comparison article will be used to some extent.

Assuming time-series data, we will prepare a data set with data of about 80,000 to 120,000 rows and 5 columns per day, and prepare data for 5 years from January 2016 to the end of December 2020.

The five columns are configured as follows.

--column_a: int-> Set a random value in the range of 0 to 4999999. --column_b: str-> Set the date and time character string (example: 2020-12-31 15:10:20). --column_c: int-> Set a random value in the range of 0 to 100. --column_d: int-> Set one of 100, 300, 500, 1000, 3000, 10000. --column_e: str-> char_num 20 Set a lowercase and uppercase alphabetic string of characters.

In addition, HDF5 has a different hierarchical structure depending on the library, so each library is saved separately.

from string import ascii_letters
import random
from datetime import date, datetime

import numpy as np
import pandas as pd
import dask.dataframe as dd
from pandas.tseries.offsets import Day
import vaex


def make_random_ascii_str(char_num):
    """
Generates a random lowercase and uppercase string with the specified number of characters.

    Parameters
    ----------
    char_num : int
The number of characters to generate.

    Returns
    -------
    random_str : str
The generated string.
    """
    return ''.join(random.choices(population=ascii_letters, k=char_num))


def make_pandas_df(row_num, char_num, date_str):
    """
Generate a Pandas data frame for validation.

    Parameters
    ----------
    row_num : int
The number of rows in the data frame to generate.
    char_num : str
The number of characters in the value to set in the string column.
    date_str : str
The string of the target date.

    Returns
    -------
    pandas_df : pd.DataFrame
The generated data frame. Each value is set in the following 5 columns.
        - column_a : int ->A random value in the range 0-4999999.
        - column_b : str ->Date and time strings (eg: 2020-12-31 15:10:20）。
        - column_c : int ->Random value in the range 0-100.
        - column_d : int -> 100, 300, 500, 1000, 3000,Any value of 10000.
        - column_e : str -> char_num A string of lowercase and uppercase alphabets with a number of 20 characters.
    """
    pandas_df = pd.DataFrame(
        index=np.arange(0, row_num), columns=['column_a'])
    row_num = len(pandas_df)
    pandas_df['column_a'] = np.random.randint(0, 5000000, size=row_num)
    
    random_hours = np.random.randint(low=0, high=24, size=row_num)
    random_minutes = np.random.randint(low=0, high=60, size=row_num)
    random_seconds = np.random.randint(low=0, high=60, size=row_num)
    times = []
    for i in range(row_num):
        hour_str = str(random_hours[i]).zfill(2)
        minute_str = str(random_minutes[i]).zfill(2)
        second_str = str(random_seconds[i]).zfill(2)
        time_str = f'{date_str} {hour_str}:{minute_str}:{second_str}'
        times.append(time_str)
    pandas_df['column_b'] = times

    pandas_df['column_c'] = np.random.randint(0, 100, size=row_num)
    pandas_df['column_d'] = np.random.choice(a=[100, 300, 500, 1000, 3000, 10000], size=row_num)
    pandas_df['column_e'] = [make_random_ascii_str(char_num=20) for _ in range(row_num)]
    pandas_df.sort_values(by='column_b', inplace=True)
    return pandas_df


def get_pandas_hdf5_file_path(date_str):
    """
Get the path of the Pandas HDF5 file for the target date.

    Parameters
    ----------
    date_str : str
The character string of the target date.
    
    Returns
    -------
    file_path : str
The generated file path.
    """
    return f'pandas_{date_str}.hdf5'


def get_dask_hdf5_file_path(date_str):
    """
Get the path of the Dask HDF5 file for the target date.

    Parameters
    ----------
    date_str : str
The character string of the target date.
    
    Returns
    -------
    file_path : str
The generated file path.
    """
    return f'dask_{date_str}.hdf5'


def get_vaex_hdf5_file_path(date_str):
    """
Get the path of the Vaex HDF5 file for the target date.

    Parameters
    ----------
    date_str : str
The character string of the target date.
    
    Returns
    -------
    file_path : str
The generated file path.
    """
    return f'vaex_{date_str}.hdf5'


def get_parquet_file_path(date_str):
    """
Get the path of the Parquet file of the target date.

    Parameters
    ----------
    date_str : str
The character string of the target date.
    
    Returns
    -------
    file_path : str
The generated file path.
    """
    return f'{date_str}.parquet'


def save_data():
    """
Save each file of time series data for verification.
    """
    current_date = date(2016, 1, 1)
    last_date = date(2020, 12, 31)
    while current_date <= last_date:
        date_str = current_date.strftime('%Y-%m-%d')
        row_num = random.randint(80_000, 120_000)
        print(
            datetime.now(), date_str, 'Start saving process. Number of lines:', row_num)
        
        pandas_df = make_pandas_df(
            row_num=row_num, char_num=20, date_str=date_str)
        vaex_df = vaex.from_pandas(df=pandas_df, copy_index=False)
        dask_df = dd.from_pandas(data=pandas_df, npartitions=1)
        
        pandas_df.to_hdf(
            path_or_buf=get_pandas_hdf5_file_path(date_str=date_str),
            key='data', mode='w')
        
        dask_df.to_hdf(
            path_or_buf=get_dask_hdf5_file_path(date_str=date_str),
            key='data', mode='w')
        
        vaex_df.export_hdf5(
            path=get_vaex_hdf5_file_path(date_str=date_str))
        
        vaex_df.export_parquet(
            path=get_parquet_file_path(date_str=date_str))
        
        current_date += Day()


save_data()

2021-01-17 07:58:24.950967 2016-01-01 save processing started. Number of lines: 83831
2021-01-17 07:58:26.994136 2016-01-02 save process started. Number of lines: 117457
2021-01-17 07:58:29.859080 2016-01-Start saving 03. Number of lines: 101470
2021-01-17 07:58:32.381448 2016-01-04 save process started. Number of lines: 88966
...

Add reading and calculation processing

Due to the lazy evaluation of Dask and Vaex, it is not possible to make a comparison only by reading, so we will proceed by assuming a comparison of how long it will take to complete the calculation in the form of reading → processing such as some calculation.

This time, we will set up the following two patterns.

-[Pattern 1]: Slice only the lines containing the character string ab with column_e-> Calculate the number of lines after slicing. -[Pattern 2]: Slice the value of column_a to a value of 3 million or less-> Slice only the line where the value of the string of column_e starts with a-> GROUP BY with the value of column_d-> column_a for each group Calculate the maximum value of.

Addition of reading process

We will add it in Pandas, Dask, and Vaex respectively.

def read_pandas_df_from_hdf5(start_date, last_date):
    """
Reads Pandas dataframes in the specified date range from HDF5 files.

    Parameters
    ----------
    start_date : date
The start date of the date range.
    last_date : date
The last day of the date range.
    
    Returns
    -------
    df : pd.DataFrame
The loaded Pandas data frame.
    """
    df_list = []
    current_date = start_date
    while current_date <= last_date:
        date_str = current_date.strftime('%Y-%m-%d')
        file_path = get_pandas_hdf5_file_path(date_str=date_str)
        df = pd.read_hdf(path_or_buf=file_path, key='data')
        df_list.append(df)
        current_date += Day()
    df = pd.concat(df_list, ignore_index=True, copy=False)
    return df


def read_dask_df_from_hdf5(start_date, last_date):
    """
Read Dask dataframes in the specified date range from HDF5 files

    Parameters
    ----------
    start_date : date
The start date of the date range.
    last_date : date
The last day of the date range.
    
    Returns
    -------
    df : dd.DataFrame
The loaded Dask data frame.
    """
    df_list = []
    current_date = start_date
    while current_date <= last_date:
        date_str = current_date.strftime('%Y-%m-%d')
        file_path = get_dask_hdf5_file_path(date_str=date_str)
        df = dd.read_hdf(pattern=file_path, key='data')
        df_list.append(df)
        current_date += Day()
    df = dd.concat(dfs=df_list)
    return df


def read_vaex_df_from_hdf5(start_date, last_date):
    """
Read Vaex dataframes in a specified date range from an HDF5 file

    Parameters
    ----------
    start_date : date
The start date of the date range.
    last_date : date
The last day of the date range.
    
    Returns
    -------
    df : vaex.dataframe.DataFrame
The loaded Vaex data frame.
    """
    file_paths = []
    current_date = start_date
    while current_date <= last_date:
        date_str = current_date.strftime('%Y-%m-%d')
        file_path = get_vaex_hdf5_file_path(date_str=date_str)
        file_paths.append(file_path)
        current_date += Day()
    vaex_df = vaex.open_many(filenames=file_paths)
    return vaex_df


def read_pandas_df_from_parquet(start_date, last_date):
    """
Read Pandas dataframes in the specified date range from Parquet.

    Parameters
    ----------
    start_date : date
The start date of the date range.
    last_date : date
The last day of the date range.
    
    Returns
    -------
    df : pd.DataFrame
The loaded Pandas data frame.
    """
    df_list = []
    current_date = start_date
    while current_date <= last_date:
        date_str = current_date.strftime('%Y-%m-%d')
        file_path = get_parquet_file_path(date_str=date_str)
        df = pd.read_parquet(path=file_path)
        df_list.append(df)
        current_date += Day()
    df = pd.concat(df_list, ignore_index=True, copy=False)
    return df


def read_dask_df_from_parquet(start_date, last_date):
    """
Read Dask dataframes in a specified date range from a Parquet file

    Parameters
    ----------
    start_date : date
The start date of the date range.
    last_date : date
The last day of the date range.
    
    Returns
    -------
    df : dd.DataFrame
The loaded Dask data frame.
    """
    df_list = []
    current_date = start_date
    while current_date <= last_date:
        date_str = current_date.strftime('%Y-%m-%d')
        file_path = get_parquet_file_path(date_str=date_str)
        df = dd.read_parquet(path=file_path)
        df_list.append(df)
        current_date += Day()
    df = dd.concat(dfs=df_list)
    return df


def read_vaex_df_from_parquet(start_date, last_date):
    """
Read Vaex dataframes in the specified date range from a Parquet file

    Parameters
    ----------
    start_date : date
The start date of the date range.
    last_date : date
The last day of the date range.
    
    Returns
    -------
    df : vaex.dataframe.DataFrame
The loaded Vaex data frame.
    """
    file_paths = []
    current_date = start_date
    while current_date <= last_date:
        date_str = current_date.strftime('%Y-%m-%d')
        file_path = get_parquet_file_path(date_str=date_str)
        file_paths.append(file_path)
        current_date += Day()
    vaex_df = vaex.open_many(filenames=file_paths)
    return vaex_df

Addition of pattern 1 processing

--Slice only lines containing the string ab with column_e --Calculate the number of rows after slicing

We will add the process in each library.

def calculate_pattern_1_with_pandas_df(pandas_df):
    """
Calculate the first pattern in a Pandas data frame.

    Parameters
    ----------
    pandas_df : pd.DataFrame
The target Pandas data frame.
    
    Returns
    -------
    row_count : int
The number of lines in the calculation result.
    """
    pandas_df = pandas_df[pandas_df['column_e'].str.contains('ab')]
    row_count = len(pandas_df)
    return row_count


def calculate_pattern_1_with_dask_df(dask_df):
    """
The calculation of the first pattern is done in the Dask data frame.

    Parameters
    ----------
    dask_df : dd.DataFrame
The data frame of the target Dask.

    Returns
    -------
    row_count : int
The number of lines in the calculation result.
    """
    dask_df = dask_df[dask_df['column_e'].str.contains('ab')]
    row_count = len(dask_df)
    return row_count


def calculate_pattern_1_with_vaex_df(vaex_df):
    """
The calculation of the third pattern is performed in the Vaex data frame.

    Parameters
    ----------
    vaex_df : vaex.dataframe.DataFrame
The target Vaex data frame.

    Returns
    -------
    row_count : int
The number of lines in the calculation result.
    """
    vaex_df = vaex_df[vaex_df['column_e'].str.contains('ab')]
    row_count = len(vaex_df)
    return row_count

Addition of pattern 2 processing

--Slice to a value of 3 million or less for column_a --Slice only to the line where the value of the string of column_e starts with a --GROUP BY with the value of column_d --Calculate the maximum value of column_a for each group

We will add the process in each library.

def calculate_pattern_2_with_pandas_df(pandas_df):
    """
Calculate the second pattern in a Pandas data frame.

    Parameters
    ----------
    pandas_df : pd.DataFrame
The target Pandas data frame.

    Returns
    -------
    max_sr : pd.Series
A series that stores each calculated total value.
    """
    pandas_df = pandas_df[pandas_df['column_a'] <= 3_000_000]
    pandas_df = pandas_df[pandas_df['column_e'].str.startswith('a')]
    grouped = pandas_df.groupby(by='column_d')
    max_df = grouped.max()
    max_sr = max_df['column_a']
    return max_sr


def calculate_pattern_2_with_dask_df(dask_df):
    """
The calculation of the second pattern is done in the Dask data frame.

    Parameters
    ----------
    dask_df : dd.DataFrame
The data frame of the target Dask.

    Returns
    -------
    max_sr : pd.Series
A series that stores each calculated total value.
    """
    dask_df = dask_df[dask_df['column_a'] <= 3_000_000]
    dask_df = dask_df[dask_df['column_e'].str.startswith('a')]
    grouped = dask_df.groupby(by='column_d')
    max_df = grouped.max()
    max_sr = max_df['column_a']
    max_sr = max_sr.compute()
    return max_sr


def calculate_pattern_2_with_vaex_df(vaex_df):
    """
The calculation of the second pattern is performed in the Vaex data frame.

    Parameters
    ----------
    vaex_df : vaex.dataframe.DataFrame
The target Vaex data frame.

    Returns
    -------
    max_sr : pd.Series
A series that stores each calculated total value.
    """
    vaex_df = vaex_df[vaex_df['column_a'] <= 3_000_000]
    vaex_df = vaex_df[vaex_df['column_e'].str.startswith('a')]
    max_df = vaex_df.groupby(
        by='column_d',
        agg={
            'column_a': vaex.agg.max,
        })
    max_df = max_df.to_pandas_df(column_names=['column_a', 'column_d'])
    max_df.index = max_df['column_d']
    max_sr = max_df['column_a']
    return max_sr

Addition of read and aggregate processing

Since each reading process and aggregation process have been added, we will write a process that combines them.

from timeit import timeit
from copy import deepcopy
import sys


class ReadAndCalcRunner:

    def __init__(self, label, pattern, read_func, calc_func):
        """
A class that handles the setting and execution processing of the combination of reading and calculation processing.

        Parameters
        ----------
        label : str
A label for identifying the target combination.
Example: csv_no_compression_pandas
        pattern : int
Target pattern (1 to 3).
        read_func : Callable
A function that handles reading processing. Arguments are optional and change the data frame
You need a format.
        calc_func : Callable
A function that handles computational processing. A format that accepts a data frame as the first argument
You need to specify something.
        """
        self.label = label
        self.pattern = pattern
        self._read_func = read_func
        self._calc_func = calc_func

    def run(self, n, start_date, last_date, debug=False):
        """
Perform reading and calculation processing. After execution, mean_in seconds attribute
The average number of seconds of execution (float) is set.

        Parameters
        ----------
        n : int
Number of executions. The larger the number, the higher the accuracy of the processing time, but it is completed.
Please note that it will take a long time to do so.
        start_date : date
The start date of the date range.
        last_date : date
The last day of the date range.
        debug : bool, default False
Debug settings. If True is specified, the calculation result will be output.
        """
        statement = 'df = self._read_func(start_date=start_date, last_date=last_date);'
        if not debug:
            statement += 'self._calc_func(df);'
        else:
            statement += 'result = self._calc_func(df); print(result)'
        result_seconds = timeit(
            stmt=statement,
            number=n, globals=locals())
        self.mean_seconds = result_seconds / n


this_module = sys.modules[__name__]
FORMATS = (
    'parquet',
    'hdf5',
)
LIBS = (
    'pandas',
    'dask',
    'vaex',
)
PATTERNS = (1, 2)

runners = []

for format_str in FORMATS:
    for lib_str in LIBS:
        for pattern in PATTERNS:
            label = f'{format_str}_{lib_str}'
            read_func_name = f'read_{lib_str}_df_from_{format_str}'
            read_func = getattr(this_module, read_func_name)

            calc_func_name = f'calculate_pattern_{pattern}_with_{lib_str}_df'
            calc_func = getattr(this_module, calc_func_name)

            runners.append(
                ReadAndCalcRunner(
                    label=label,
                    pattern=pattern,
                    read_func=read_func,
                    calc_func=calc_func)
            )

Added processing for plotting

We will add functions for comparing processing times.

import matplotlib
from matplotlib.ticker import ScalarFormatter, FormatStrFormatter
import matplotlib.pyplot as plt

matplotlib.style.use('ggplot')


def plot_time_info(time_sr, patten, print_sr = True):
    """
Plot the processing time.

    Parameters
    ----------
    time_sr : pd.Series
A series that stores the processing time.
    patten : int
Target aggregation processing pattern (1 to 3).
    print_sr : bool, default True
Whether to output and display a series of plot contents.
    """
    sr = time_sr.copy()
    sr.sort_values(inplace=True, ascending=False)
    if print_sr:
        print(sr.sort_values(ascending=True))
    title = f'Read and calculation seconds (pattern: {patten})'
    ax = sr.plot(kind='barh', figsize=(10, len(sr) // 2), title=title)
    ax.xaxis.set_major_formatter(FormatStrFormatter('%.3f'))
    plt.show()

Addition of processing to flow each processing

In order to simplify the description, add a function to process each process (reading, aggregation, visualization) at once.

def run_and_get_result_time_sr(
        runners, pattern, n, start_date, last_date, skip_pandas, skip_dask_hdf5,
        skip_dask_parquet):
    """
Executed each measurement process of the specified pattern and stored the value of each number of seconds of the result.
Get the series.

    Parameters
    ----------
    runners : list of ReadAndCalcRunner
A list of instances that hold definitions of execution processes.
    pattern : int
Pattern to execute(1～3)。
    n : int
Number of executions. The larger the number, the higher the accuracy of the processing time, but it is completed.
Please note that it will take a long time to do so.
    start_date : date
The start date of the date range.
    last_date : date
The last day of the date range.
    skip_pandas : bool
Whether to skip Pandas processing.
    skip_dask_hdf5 : bool
Whether to skip processing of HDF5 format Dask.
    skip_dask_parquet : bool
Whether to skip the processing of Dask in Parquet format.

    Returns
    -------
    sr : pd.Series
A series that stores measurement results. Each format and index
A concatenated string of labels and patterns for library identification is set,
The value is set to the number of seconds.
    """
    runners = deepcopy(runners)
    data_dict = {}
    for runner in runners:
        if runner.pattern != pattern:
            continue
        if skip_pandas and 'pandas' in runner.label:
            continue
        if skip_dask_hdf5 and 'dask' in runner.label and 'hdf5' in runner.label:
            continue
        if skip_dask_parquet and 'dask' in runner.label and 'parquet' in runner.label:
            continue
        label = f'{runner.label}_{pattern}'
        print(datetime.now(), label, 'Start processing...')
        runner.run(n=n, start_date=start_date, last_date=last_date)
        data_dict[label] = runner.mean_seconds
    sr = pd.Series(data=data_dict)
    return sr


def run_overall(
        n, pattern, start_date, last_date, skip_pandas, skip_dask_hdf5,
        skip_dask_parquet):
    """
Each process (reading, aggregation, visualization, etc.) for the target pattern
Run overall.

    Parameters
    ----------
    n : int
The number of reads and calculations performed. The larger the number, the more accurate the processing time,
Please note that it will take a long time to complete.
    pattern : int
The target calculation pattern (1 to 3).
    start_date : date
The start date of the date range.
    last_date : date
The last day of the date range.
    skip_pandas : bool
Whether to skip Pandas processing.
    skip_dask_hdf5 : bool
Whether to skip processing of HDF5 format Dask.
    skip_dask_parquet : bool
Whether to skip the processing of Dask in Parquet format.
    """
    time_sr = run_and_get_result_time_sr(
        runners=runners, pattern=pattern, n=n, start_date=start_date,
        last_date=last_date, skip_pandas=skip_pandas,
        skip_dask_hdf5=skip_dask_hdf5,
        skip_dask_parquet=skip_dask_parquet)
    plot_time_info(time_sr=time_sr, patten=pattern)

Actually execute

Now that we are ready, we will add up each processing time. Try each of the following conditions. In addition, Pandas targets up to 3 months, and after that, it will be executed only in Dask and Vaex.

――One month (about 3 million lines) ――For 3 months (about 9 million lines) ――For 6 months (about 18 million lines) --One year's worth (about 36 million lines) ――For 3 years (about 108 million lines) ――For 5 years (about 180 million lines)

Aggregation for one month

pattern 1:

run_overall(
    n=5, pattern=1,
    start_date=date(2016, 1, 1),
    last_date=date(2016, 1, 31),
    skip_pandas=False)

hdf5_vaex_1         1.015516
parquet_vaex_1      1.028920
parquet_pandas_1    2.685143
parquet_dask_1      2.828006
hdf5_pandas_1       3.311069
hdf5_dask_1         7.616159

--Although it is Snappy compressed, the processing time of Parquet + Vaex is quite close to that of uncompressed HDF5. --Pandas and Dask seem to be slower in HDF5 (despite uncompressed). I'm particularly concerned about Dask's processing time.

Pattern 2:

run_overall(
    n=5, pattern=2,
    start_date=date(2016, 1, 1),
    last_date=date(2016, 1, 31),
    skip_pandas=False)

hdf5_vaex_2         0.766808
parquet_vaex_2      0.848183
parquet_pandas_2    2.436566
parquet_dask_2      2.961728
hdf5_pandas_2       4.134251
hdf5_dask_2         8.657277

Aggregation for 3 months

pattern 1:

run_overall(
    n=5, pattern=1,
    start_date=date(2016, 1, 1),
    last_date=date(2016, 3, 31),
    skip_pandas=False)

hdf5_vaex_1          2.260799
parquet_vaex_1       2.649166
parquet_dask_1       8.578201
parquet_pandas_1     8.656629
hdf5_pandas_1        9.994132
hdf5_dask_1         22.766739

Pattern 2:

run_overall(
    n=5, pattern=2,
    start_date=date(2016, 1, 1),
    last_date=date(2016, 3, 31),
    skip_pandas=False)

hdf5_vaex_2          1.894970
parquet_vaex_2       2.529330
parquet_pandas_2     7.110901
hdf5_pandas_2        9.252198
parquet_dask_2      10.688318
hdf5_dask_2         23.362928

Aggregation for 6 months

From here, skip Pandas and proceed only with Dask and Vaex. Also, due to processing time, we will reduce the number of executions for each pattern from 5 to 3 times.

In addition, Dask + HDF5 is quite slow, so I'll adjust it so that it can be skipped due to processing time.

pattern 1:

run_overall(
    n=3, pattern=1,
    start_date=date(2016, 1, 1),
    last_date=date(2016, 6, 30),
    skip_pandas=True,
    skip_dask_hdf5=True)

hdf5_vaex_1        4.621812
parquet_vaex_1     5.633019
parquet_dask_1    17.827765

Pattern 2:

run_overall(
    n=3, pattern=2,
    start_date=date(2016, 1, 1),
    last_date=date(2016, 6, 30),
    skip_pandas=True,
    skip_dask_hdf5=True)

hdf5_vaex_2        4.231214
parquet_vaex_2     5.312496
parquet_dask_2    17.153308

Aggregation for one year

We will reduce the number of executions to 2.

pattern 1:

run_overall(
    n=2, pattern=1,
    start_date=date(2016, 1, 1),
    last_date=date(2016, 12, 31),
    skip_pandas=True,
    skip_dask_hdf5=True)

hdf5_vaex_1        9.618381
parquet_vaex_1    11.091080
parquet_dask_1    36.335810

Pattern 2:

run_overall(
    n=2, pattern=2,
    start_date=date(2016, 1, 1),
    last_date=date(2016, 12, 31),
    skip_pandas=True,
    skip_dask_hdf5=True)

hdf5_vaex_2        9.132881
parquet_vaex_2    11.136143
parquet_dask_2    34.377085

Aggregation for 3 years

The number of trials will be greatly reduced, but after that, due to processing time, it will be executed only once.

pattern 1:

run_overall(
    n=1, pattern=1,
    start_date=date(2016, 1, 1),
    last_date=date(2018, 12, 31),
    skip_pandas=True,
    skip_dask_hdf5=True)

hdf5_vaex_1        40.676083
parquet_vaex_1     43.035784
parquet_dask_1    100.698389

Pattern 2:

run_overall(
    n=1, pattern=2,
    start_date=date(2016, 1, 1),
    last_date=date(2018, 12, 31),
    skip_pandas=True,
    skip_dask_hdf5=True)

hdf5_vaex_2        40.061167
parquet_vaex_2     42.218093
parquet_dask_2    102.830116

Aggregation for 5 years

Is it affected by running on Docker? For some reason, the Jupyter kernel crashes while running Dask's, so I'll try to proceed with Vaex only for 5 years.

pattern 1:

run_overall(
    n=1, pattern=1,
    start_date=date(2016, 1, 1),
    last_date=date(2020, 12, 31),
    skip_pandas=True,
    skip_dask_hdf5=True,
    skip_dask_parquet=True)

parquet_vaex_1    95.578259
hdf5_vaex_1       99.258315

Since the number of trials is one, there is a possibility that it may have shaken, but the result is faster with Parquet.

Pattern 2:

run_overall(
    n=1, pattern=2,
    start_date=date(2016, 1, 1),
    last_date=date(2020, 12, 31),
    skip_pandas=True,
    skip_dask_hdf5=True,
    skip_dask_parquet=True)

hdf5_vaex_2       78.231278
parquet_vaex_2    93.696743

Here, HDF5 is usually faster.

[PYTHON] Performance comparison of Parquet, Vaex, Dask, etc. in HDF5 with multiple files

Environment to use

Target format

Data preparation

Add reading and calculation processing

Addition of reading process

Addition of pattern 1 processing

Addition of pattern 2 processing

Addition of read and aggregate processing

Added processing for plotting

Addition of processing to flow each processing

Actually execute

Aggregation for one month

Aggregation for 3 months

Aggregation for 6 months

Aggregation for one year

Aggregation for 3 years

Aggregation for 5 years