[Python] Values differ on multiple reads from parquet files using polars read_parquet but not...

I read data from the same parquet files multiple times using polars (polars rust engine and pyarrow) and using pandas pyarrow backend (not fastparquet as it was very slow), see below code.

All the parquetfiles contain a column called "backscat" where every element is a list of 1.5 Million floating points (it is a time series).

If the hash of the dataframe is different, my script checks the differences in the values of the "backscat" column and if there are values outside "rtol = 1e-7, atol = 1e-10" saved them as "violations".

I let that script run on two workstations with an identical cpu, with the identical virtual environment (poetry) and python version (3.12.8) and polars Version: 1.20.0, Ubuntu 22.04. and they also have exactly the same memory configuration.

On workstation A my analysis show that there are significant differences between file reads on the same file. For example, sorting for the largest absolute differences, for the polars rust backend, the 5 biggest values are:

name_test data_1 data_2 abs_diff abs_relative_diff
polars_rust -3.041666 -38.666656 35.62499 11.712328
polars_rust -38.666656 -3.041666 35.62499 0.921336
polars_rust -2.927914 -27.423315 24.495401 8.36616
polars_rust -27.423315 -2.927914 24.495401 0.893233
polars_rust -2.927876 -27.42301 24.495134 8.366178

As you can see the differnces are absolutely significant, one time it reads -3.041666, the other time it reads -38.666656! On the same workstation A I do not get any violations using pandas.

Now running the same script on workstation B, with the same cpu, I do not get a single violation, not for polars, not for pandas.

Is this a known issue? Is my RAM on workstation A corrupted? But then why does this problem only occure with polars?

I now also run it in a Docker container on workstation A, with polars version 1.24.0, and the results were the same.

I observed that if a colleague runs simmulations at the same time, using all processors, I am not getting violations! Hence if polars can not use multiple processes in parallel I am not getting the error?

# CODE TO READ PARQUET FILES AND DETECT DIFFERENCES
import hashlib
from pathlib import Path
from typing import Any, Callable

import numpy as np
import pandas as pd
import polars as pl
from tqdm import tqdm


def hash_dataframe_polars(df):
return hashlib.sha256(df.select("backscat").explode("backscat").write_csv(None).encode()).hexdigest()


def hash_dataframe_pandas(df):
return hashlib.sha256(df["backscat"].to_csv().encode()).hexdigest()


def find_precision_violations(
paths_parquet_files: list[Path],
name_test: str,
rtol: float,
atol: float,
hash_fun: Callable[[Any], str],
read_parquet_file: Callable[[Path], Any],
num_reps_per_file: int,
output_dir: Path = Path("./results_check_ram_or_cpu_catch"), # Default output directory
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
precision_violated = []
errors_reading_files = []

# Ensure output directory exists
output_dir.mkdir(parents=True, exist_ok=True)

# Use tqdm to monitor file processing
for file_index, path_file in enumerate(tqdm(paths_parquet_files, desc=f"Processing {name_test} Files")):
previous_hash = None
previous_df = None

# Use tqdm to monitor repetitions per file
for rep in tqdm(range(num_reps_per_file), desc=f"Reps for File {file_index}", leave=False):
try:
df = read_parquet_file(path_file)
except Exception as e:
errors_reading_files.append(
{"name_test": name_test, "file_index": file_index, "rep": rep, "error": str(e)}
)
continue

hash = hash_fun(df)
if hash != previous_hash and previous_hash is not None:
tqdm.write(f"{name_test}, hashes differ for file {path_file}")

for i in range(df.shape[0]):
data_1 = np.asarray(df["backscat"])
data_2 = np.asarray(previous_df["backscat"]) # type: ignore
mask = ~np.isclose(data_1, data_2, rtol=rtol, atol=atol)

if np.any(mask):
print(f"Precision violation found for {sum(mask)} elements")

precision_violated.append(
{
"name_test": name_test,
"data_1": data_1[mask].tolist(),
"data_2": data_2[mask].tolist(),
"i": i,
"rep": rep,
"file_index": file_index,
}
)

previous_hash = hash
previous_df = df

# Convert results to Polars DataFrame and save as Parquet
if precision_violated:
df_precision_violated = pl.DataFrame(precision_violated)
df_precision_violated.write_parquet(output_dir / f"precision_violations_{name_test}.parquet")

if errors_reading_files:
df_errors = pl.DataFrame(errors_reading_files)
df_errors.write_parquet(output_dir / f"errors_reading_files_{name_test}.parquet")

return precision_violated, errors_reading_files


list_set_up = [
{
"name_test": "polars_rust",
"read_parquet_file": lambda path_file: pl.read_parquet(
path_file, columns=["date", "backscat"], use_pyarrow=False
),
"hash_fun": hash_dataframe_polars,
},
{
"name_test": "pandas_pyarrow",
"read_parquet_file": lambda path_file: pd.read_parquet(
path_file, columns=["date", "backscat"], engine="pyarrow"
),
"hash_fun": hash_dataframe_pandas,
},
# {
# "name_test": "pandas_fastparquet",
# "read_parquet_file": lambda path_file: pd.read_parquet(
# path_file, columns=["date", "backscat"], engine="fastparquet"
# ),
# "hash_fun": hash_dataframe_pandas,
# },
{
"name_test": "polars_pyarrow",
"read_parquet_file": lambda path_file: pl.read_parquet(
path_file, columns=["date", "backscat"], use_pyarrow=True
),
"hash_fun": hash_dataframe_polars,
},
]

dir_parquet_files = Path("cache/212_artificial_data_set/parquet_files")
paths_parquet_files = list(dir_parquet_files.glob("*.parquet"))
rtol = 1e-7
atol = 1e-10
num_reps_per_file = 10
all_violations = []
all_errors_reading_files = []
for set_up in list_set_up:
violations, errors_reading_files = find_precision_violations(
paths_parquet_files=paths_parquet_files,
name_test=set_up["name_test"],
rtol=rtol,
atol=atol,
hash_fun=set_up["hash_fun"],
read_parquet_file=set_up["read_parquet_file"],
num_reps_per_file=num_reps_per_file,
)

all_violations += violations
all_errors_reading_files += errors_reading_files


# %% save violations to csv save errors reading files to csv
violations_df = pd.DataFrame(all_violations)
print(violations_df)
violations_df.to_csv("all_violations.csv")

errors_reading_files_df = pd.DataFrame(all_errors_reading_files)
print(errors_reading_files_df)
errors_reading_files_df.to_csv("all_errors_reading_files.csv")


# ANALYSIS SCRIPT
import ast

import matplotlib.pyplot as plt
import polars as pl

violations_df = pl.read_csv("all_violations.csv")
errors_reading_files_df = pl.read_csv("all_errors_reading_files.csv")

violations_df = pl.read_csv("all_violations.csv")

violations_df = violations_df.with_columns(
pl.col("data_1").map_elements(lambda x: ast.literal_eval(x), return_dtype=pl.List(pl.Float64)),
pl.col("data_2").map_elements(lambda x: ast.literal_eval(x), return_dtype=pl.List(pl.Float64)),
)

violations_df = violations_df.with_columns(
pl.col("data_1").len().alias("len_data_1"),
pl.col("data_2").len().alias("len_data_2"),
)

if violations_df.filter(pl.col("len_data_1") != pl.col("len_data_2")).height > 0:
print("Lengths differ")

violations_df = violations_df.drop("len_data_1").rename({"len_data_2": "num_violations"})

tot_violations = (
violations_df.group_by("name_test")
.agg(pl.col("num_violations").sum().alias("total_violations"))
.sort("total_violations")
)

print(tot_violations)

# %%
for name_test in violations_df["name_test"].unique():
df = violations_df.filter(pl.col("name_test") == name_test)
df = df.explode(["data_1", "data_2"])

df = df.with_columns((pl.col("data_1") - pl.col("data_2")).alias("diff"))
df = df.with_columns(pl.col("diff").abs().alias("abs_diff"))
df = df.with_columns((pl.col("abs_diff") / pl.col("data_1").abs()).alias("abs_relative_diff"))

cols = ["name_test", "data_1", "data_2", "abs_diff", "abs_relative_diff"]
df = df.sort(by="abs_diff", descending=True)
print("sorted by abs_diff")
print("head")
print(df.select(cols).head(10))
print("tail")
print(df.select(cols).tail(10))

df = df.sort(by="abs_relative_diff", descending=True)
print("sorted by abs_relative_diff")
print("head")
print(df.select(cols).head(10))
print("tail")
print(df.select(cols).tail(10))

Continue reading...