Hourly data pipeline

[qwert666]

Hi

I'm trying to use Popmon for monitoring changes in my inventory, and can't make it run properly in my setup.

My setup consists of a pipeline that process a lot of data and I would like to generate profile and comparison metrics based on historical histograms that I've stored.

With each run I would have only access to a dataframe that consists of one hour of data and I would then compare it with the pre-computed hourly histograms, that I've beforehand precomputed and stored

hourly_histograms = {}
features = [
   "revenue",
   "impressions"
]

for hour in range(0, 24):
    df_hour = df.where(F.hour(F.col("datetime")) == hour).select(
        F.log("revenue").alias("revenue"),
        F.log("impressions").alias("impressions")
    )
    hourly_histogram = make_histograms(df_hour, features=features)
    hourly_histograms[hour] = histogram
    # I would serialize and store it as well  

When running:

current_hour = 1
report = new_hour_df.pm_stability_report(
    settings=settings=Settings(
        reference_type="external"
    ),
    reference=hourly_histograms[current_hour]
)

> [log] Time-axis automatically set to "datetime"
> [Exception] RuntimeError: Specified time-axis "datetime" not found in dataframe.

So I assume that some time-axis is needed so when I've tried to use datetime then I'm getting:

report.datastore['hists']['revenue'].n_bins #=> 1

which also looks weird? is it possible to use popmon in my setup? don't know how to properly validate this setup

my data:

> new_hour_df.select(F.min("datetime"), F.max("datetime"), F.count("datetime")).show()

+-------------------+-------------------+---------------+
|      min(datetime)|      max(datetime)|count(datetime)|
+-------------------+-------------------+---------------+
|2023-05-11 01:00:00|2023-05-11 01:00:00|           9709|
+-------------------+-------------------+---------------+

> df_hour.select(F.min("datetime"), F.max("datetime"), F.count("datetime")).show()

+-------------------+-------------------+---------------+
|      min(datetime)|      max(datetime)|count(datetime)|
+-------------------+-------------------+---------------+
|2023-04-28 01:00:00|2023-05-10 01:00:00|         130379|
+-------------------+-------------------+---------------+

[qwert666]

I found some examples where instead of using a time domain related column an index column is created

so in my case it would be

df = df.withColumn("index", F.monotonically_increasing_id())

settings=Settings(
    reference_type="external",
    time_axis="index"
)

features = [
   "index:revenue",
   "index:impressions"
]

would this be the proper way of approaching the problem and how would I interpret the profiles and comparisons then, seems a bit confusing 😕

[mbaak]

Hi,
Did you go through this example on incremental datasets?
https://nbviewer.org/github/ing-bank/popmon/blob/master/popmon/notebooks/popmon_tutorial_incremental_data.ipynb
The stitching of (hourly) histograms discussed there makes a time axis for you that popmon then picks up for its reporting.

Let me know if that solves your problem.

Best, -Max

[qwert666]

Yes, I've went thru all the examples but it's not clear, I'm definitely getting a wrong report, I'll just put a full snippet:

hist_list = []

features = [
    'datetime:revenue',
    'datetime:impressions'
]

df = spark.read.format("parquet").load("./data.parquet").where(
    F.col("datetime").cast("date").isin(
        datetime.datetime(2023,5,8),
        datetime.datetime(2023,5,9),
        datetime.datetime(2023,5,10)
    )
)

df = df.select(
    F.log("revenue").alias("revenue"),
    F.log("impressions").alias("impressions"),
    "datetime"  # datetime is always with full hours e.g. 2023-05-19 20:00:00, 2023-05-19 21:00:00
)

# defining a binning specification based on the full dataset? this would need to be serialized and stored for future runs?
bin_spec = get_bin_specs(
    df.pm_make_histograms(
        features=features, time_axis="datetime", time_width="1h"
    )
)

# filtering out the most recent data and is used in the example as my delta
last_datetime = df.select(F.max("datetime").alias("max_datetime")).collect()[0]["max_datetime"]
# > datetime.datetime(2023, 5, 10, 23, 0)

delta_hour_df = df.where(F.col("datetime") == last_datetime)
hist_df = df.where(F.col("datetime") != last_datetime)

datetimes = set([row['datetime'] for row in hist_df.select("datetime").collect()])

# generate histograms per datetime (hour), each datetime uses the same (hourly) binning specifications
for datetime in datetimes:
    df_hour = hist_df.where(F.col("datetime") == datetime)
    h = df_hour.pm_make_histograms(features=features, time_axis="datetime", time_width="1h", bin_specs=bin_spec)
    hist_list.append(h)
    
    
hists_delta = delta_hour_df.pm_make_histograms(
    features=features,
    time_axis="datetime",
    time_width="1h",
    bin_specs=bin_spec
)

hists4 = popmon.stitch_histograms(
    hists_basis=hist_list,
    hists_delta=hists_delta,
    time_axis="datetime"
)

report = popmon.stability_report(
     hists4,
     time_width="1h",
     settings=Settings(
         reference_type="self",
         time_axis="datetime"
     ),
)

what I'm getting is a weird time range:

time bins: 1
time range: 2023-05-10 21:30:00 - 2023-05-10 21:30:00

I would expect to get a full hour as that's what I have in my dataframe so datetime.datetime(2023, 5, 10, 23, 0)

would expect something like:
2023-05-10 23:00:00 <- compared with -> 2023-05-09 23:00:00, 2023-05-08 23:00:00
2023-05-10 22:00:00 <- compared with -> 2023-05-09 22:00:00, 2023-05-08 22:00:00

Thanks 🙏