def _cache_df_or_retrieve_cache_data_url(df, parent_cache_dir_url,
                                         parquet_row_group_size_bytes,
                                         compression_codec,
                                         dtype):
    """Check whether the df is cached.

    If so, return the existing cache file path.
    If not, cache the df into the cache_dir in parquet format and return the
    cache file path.
    Use atexit to delete the cache before the python interpreter exits.
    :param df: A :class:`pyspark.sql.DataFrame` object.
    :param parquet_row_group_size_bytes: An int denoting the number of bytes
        in a parquet row group.
    :param compression_codec: Specify compression codec.
    :param dtype: ``None``, 'float32' or 'float64', specifying the precision of the floating-point
        elements in the output dataset. Integer types will remain unchanged. If ``None``, all types
        will remain unchanged. Default 'float32'.
    :return: A string denoting the path of the saved parquet file.
    """
    # TODO
    #  Improve the cache list by hash table (Note we need use hash(df_plan +
    #  row_group_size)
    with _cache_df_meta_list_lock:
        df_plan = _get_df_plan(df)
        for meta in _cache_df_meta_list:
            if meta.row_group_size == parquet_row_group_size_bytes and \
                    meta.compression_codec == compression_codec and \
                    meta.df_plan.sameResult(df_plan) and \
                    meta.dtype == dtype and \
                    meta.parent_cache_dir_url == parent_cache_dir_url:
                return meta.cache_dir_url
        # do not find cached dataframe, start materializing.
        cached_df_meta = CachedDataFrameMeta.create_cached_dataframe_meta(
            df, parent_cache_dir_url, parquet_row_group_size_bytes,
            compression_codec, dtype)
        _cache_df_meta_list.append(cached_df_meta)
        return cached_df_meta.cache_dir_url
Check whether the df is cached.

    If so, return the existing cache file path.
    If not, cache the df into the cache_dir in parquet format and return the
    cache file path.
    Use atexit to delete the cache before the python interpreter exits.
    :param df: A :class:`pyspark.sql.DataFrame` object.
    :param parquet_row_group_size_bytes: An int denoting the number of bytes
        in a parquet row group.
    :param compression_codec: Specify compression codec.
    :param dtype: ``None``, 'float32' or 'float64', specifying the precision of the floating-point
        elements in the output dataset. Integer types will remain unchanged. If ``None``, all types
        will remain unchanged. Default 'float32'.
    :return: A string denoting the path of the saved parquet file.
    
petastorm/spark/spark_dataset_converter.py