Usage of ValueCachingWrapper in Neuraxle.

This demonstrates how you can use value caching in a Neuraxle pipeline.

Out:

Pipeline with ValueCachingWrapper
execution time: 0.22718000411987305 seconds
Pipeline without value caching
execution time: 0.7250409126281738 seconds

import os
import shutil
import time

import numpy as np

from neuraxle.pipeline import Pipeline
from neuraxle.steps.caching import PickleValueCachingWrapper
from neuraxle.steps.loop import ForEachDataInput
from neuraxle.steps.misc import Sleep
from neuraxle.steps.numpy import MultiplyByN


def main():
    value_caching_folder = 'value_caching'
    if not os.path.exists(value_caching_folder):
        os.makedirs(value_caching_folder)

    data_inputs = list(range(100))

    sleep_time = 0.001
    a = time.time()
    for i in range(5):
        p = Pipeline([
            PickleValueCachingWrapper(
                ForEachDataInput(Pipeline([Sleep(sleep_time=sleep_time), MultiplyByN(2)])),
                cache_folder=value_caching_folder
            )
        ])
        outputs_value_caching = p.transform(data_inputs)
    b = time.time()
    time_value_caching_pipeline = b - a
    print('Pipeline with ValueCachingWrapper')
    print('execution time: {} seconds'.format(time_value_caching_pipeline))

    a = time.time()
    for i in range(5):
        p = Pipeline([
            ForEachDataInput(Pipeline([Sleep(sleep_time=sleep_time), MultiplyByN(2)])),
        ])

        outputs_vanilla = p.transform(data_inputs)
    b = time.time()
    time_vanilla_pipeline = b - a
    print('Pipeline without value caching')
    print('execution time: {} seconds'.format(time_vanilla_pipeline))

    shutil.rmtree(value_caching_folder)

    assert np.array_equal(outputs_value_caching, outputs_vanilla)
    assert time_value_caching_pipeline < time_vanilla_pipeline


if __name__ == '__main__':
    main()

Total running time of the script: ( 0 minutes 1.012 seconds)

Gallery generated by Sphinx-Gallery