a31ae1708b62413c94ffae9fb6aab21b37f78212,petastorm/spark/spark_dataset_converter.py,,_cache_df_or_retrieve_cache_path,#Any#Any#Any#Any#,70

Before Change

    :param compression_codec: Specify compression codec.
    :return:          A string denoting the path of the saved parquet file.
    
    uuid_str = str(uuid.uuid4())
    save_to_dir = os.path.join(cache_dir, uuid_str)

    df.write \
        .option("compression", compression_codec) \
        .option("parquet.block.size", row_group_size) \

After Change

    with _cache_df_meta_list_lock:
        df_plan = _get_df_plan(df)
        for meta in _cache_df_meta_list:
            if meta.row_group_size == row_group_size and \
                    meta.compression_codec == compression_codec and \
                    meta.df_plan.sameResult(df_plan):
                return meta.data_path
        // do not find cached dataframe, start materializing.
        cached_df_meta = CachedDataFrameMeta.create_cached_dataframe(
            df, parent_cache_dir, row_group_size, compression_codec)
        _cache_df_meta_list.append(cached_df_meta)
        return cached_df_meta.data_path

Instances