Robert Esteves
03/31/2026, 9:01 PMMarvin
03/31/2026, 9:01 PMMarvin
03/31/2026, 9:05 PMimport asyncio
import datetime as dt
from zoneinfo import ZoneInfo
import pandas as pd
import matplotlib.pyplot as plt
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterStartTime
from prefect.client.schemas.sorting import FlowRunSort
# Settings
DAYS_BACK = 30
TZ = ZoneInfo("America/New_York") # change to your local TZ, e.g., ZoneInfo("UTC")
async def fetch_runs():
since = dt.datetime.now(dt.timezone.utc) - dt.timedelta(days=DAYS_BACK)
fr_filter = FlowRunFilter(
start_time=FlowRunFilterStartTime(after_=since)
)
runs = []
limit = 200
offset = 0
async with get_client() as client:
while True:
batch = await client.read_flow_runs(
flow_run_filter=fr_filter,
sort=FlowRunSort.START_TIME_ASC,
limit=limit,
offset=offset,
)
if not batch:
break
runs.extend(batch)
offset += limit
return runs
def main():
runs = asyncio.run(fetch_runs())
if not runs:
print("No runs found.")
return
# Build a DataFrame
rows = []
for r in runs:
if r.start_time:
rows.append({
"flow_run_id": str(r.id),
"name": r.name,
"deployment_id": str(r.deployment_id) if r.deployment_id else None,
"state_type": r.state.type.value if r.state else None,
"start_time_utc": pd.to_datetime(r.start_time),
"end_time_utc": pd.to_datetime(r.end_time) if r.end_time else None,
})
df = pd.DataFrame(rows)
if df.empty:
print("No runs with start_time.")
return
# Convert to your local time zone
df["start_time_local"] = df["start_time_utc"].dt.tz_convert(TZ)
df["hour_of_day"] = df["start_time_local"].dt.hour
df["dow"] = df["start_time_local"].dt.dayofweek # 0=Mon, 6=Sun
df["date"] = df["start_time_local"].dt.date
# Save CSV
df.to_csv("prefect_flow_runs_last_30d.csv", index=False)
print("Wrote prefect_flow_runs_last_30d.csv")
# Histogram of starts by hour-of-day
ax = df["hour_of_day"].value_counts().sort_index().plot(kind="bar", figsize=(10,4))
ax.set_title("Flow starts by hour of day")
ax.set_xlabel("Hour of day")
ax.set_ylabel("Run count")
plt.tight_layout()
plt.savefig("prefect_run_hours.png", dpi=150)
plt.close()
print("Wrote prefect_run_hours.png")
# Heatmap: day-of-week vs hour-of-day
pivot = df.pivot_table(index="dow", columns="hour_of_day", values="flow_run_id", aggfunc="count", fill_value=0)
plt.figure(figsize=(12,4))
plt.imshow(pivot.values, aspect="auto", cmap="Blues")
plt.colorbar(label="Run count")
plt.yticks(range(7), ["Mon","Tue","Wed","Thu","Fri","Sat","Sun"])
plt.xticks(range(24), range(24))
plt.title("Flow starts heatmap (DOW x Hour)")
plt.xlabel("Hour of day")
plt.ylabel("Day of week")
plt.tight_layout()
plt.savefig("prefect_run_heatmap.png", dpi=150)
plt.close()
print("Wrote prefect_run_heatmap.png")
if __name__ == "__main__":
main()
Notes:
- Adjust TZ to your local timezone.
- Add filters for specific deployments/tags if needed by extending the FlowRunFilter (e.g., tags, deployment_id).
- This will create CSV + two PNGs you can download and share.Marvin
03/31/2026, 9:05 PMimport asyncio
import datetime as dt
from zoneinfo import ZoneInfo
import pandas as pd
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterStartTime
TZ = ZoneInfo("America/New_York")
DAYS_BACK = 30
INTERVAL_SECONDS = 3600 # 1 hour
async def fetch_hourly_history():
history_start = dt.datetime.now(dt.timezone.utc) - dt.timedelta(days=DAYS_BACK)
history_end = dt.datetime.now(dt.timezone.utc)
# Optional: additional filters (tags, deployments, etc.)
fr_filter = FlowRunFilter(
start_time=FlowRunFilterStartTime(after_=history_start, before_=history_end)
)
async with get_client() as client:
resp = await <http://client._client.post|client._client.post>(
"/flow_runs/history",
json={
"history_start": history_start.isoformat(),
"history_end": history_end.isoformat(),
"history_interval_seconds": INTERVAL_SECONDS,
"flow_runs": fr_filter.model_dump(mode="json", exclude_unset=True),
},
)
data = resp.json()
# Build a DataFrame of total starts per interval (sum across states)
rows = []
for bucket in data:
start = pd.to_datetime(bucket["interval_start"], utc=True).tz_convert(TZ)
total = sum(s.get("count_runs", 0) for s in bucket.get("states", []))
rows.append({"interval_start": start, "total_runs": total})
return pd.DataFrame(rows)
def main():
df = asyncio.run(fetch_hourly_history())
if df.empty:
print("No history data returned.")
return
# Derive hour-of-day distribution
df["hour_of_day"] = df["interval_start"].dt.hour
hourly = df.groupby("hour_of_day")["total_runs"].sum()
ax = hourly.plot(kind="bar", figsize=(10,4), title="Starts per hour (aggregated via history)")
ax.set_xlabel("Hour of day")
ax.set_ylabel("Run count")
ax.figure.tight_layout()
ax.figure.savefig("prefect_run_hours_from_history.png", dpi=150)
print("Wrote prefect_run_hours_from_history.png")
if __name__ == "__main__":
main()
Optional — Show the visual inside Prefect
- If you can host the image (e.g., on S3) at a URL, you can attach it to the run as an image artifact so it appears in the UI:
- Upload the PNG to a public or signed URL (e.g., S3 presigned URL)
- Then create an image artifact that points to that URL:
from prefect.artifacts import create_image_artifact
create_image_artifact(
image_url="<https://your-bucket.s3.amazonaws.com/prefect_run_hours.png>",
key="run-hours-visual",
description="Starts by hour-of-day for the last 30 days"
)
Would you prefer:
- all flows vs a specific deployment/tag?
- a simple histogram, or a day-of-week x hour heatmap?
- and what timezone should we use?
If you share that, I can tailor the snippet exactly to your case.