perf(weather): batch ASOS observation upserts into a single insert_all
IemClient.fetch_asos returns 24-288 rows per station per call, each previously triggering an individual UPDATE-conflict round-trip via Weather.upsert_surface_observation/2. On the Turing Pi 2 Postgres node that per-row latency dominates ingestion time. Add Weather.upsert_surface_observations/2, a bulk variant that collapses the rows into a single Repo.insert_all with the same update-only-when-changed on_conflict predicate and (station_id, observed_at) conflict target. Switch WeatherFetchWorker to use it.
This commit is contained in:
parent
ef54125aa6
commit
5255a6ba63
3 changed files with 162 additions and 5 deletions
|
|
@ -86,6 +86,94 @@ defmodule Microwaveprop.Weather do
|
|||
)
|
||||
end
|
||||
|
||||
@doc """
|
||||
Bulk-upsert surface observations for a single station via one
|
||||
`Repo.insert_all` round-trip. ASOS fetches return 24–288 rows per
|
||||
station per call — collapsing them into a single statement avoids the
|
||||
per-row UPDATE-conflict round-trip of `upsert_surface_observation/2`,
|
||||
which is expensive against the Turing Pi 2 Postgres node.
|
||||
|
||||
Rows missing `observed_at` are dropped (they can't satisfy the
|
||||
`(station_id, observed_at)` unique index). Returns the
|
||||
`{count, nil}` tuple from `Repo.insert_all/3`; `count` reflects
|
||||
affected rows (new + updated-when-changed). Returns `{0, nil}` for
|
||||
an empty input without touching the DB.
|
||||
"""
|
||||
@spec upsert_surface_observations(Station.t(), [map()]) :: {non_neg_integer(), nil}
|
||||
def upsert_surface_observations(%Station{} = station, rows) when is_list(rows) do
|
||||
now = DateTime.truncate(DateTime.utc_now(), :second)
|
||||
|
||||
entries =
|
||||
rows
|
||||
|> Enum.filter(&row_has_observed_at?/1)
|
||||
|> Enum.map(&surface_observation_entry(&1, station.id, now))
|
||||
|
||||
case entries do
|
||||
[] ->
|
||||
{0, nil}
|
||||
|
||||
_ ->
|
||||
Repo.insert_all(SurfaceObservation, entries,
|
||||
on_conflict:
|
||||
from(s in SurfaceObservation,
|
||||
update: [
|
||||
set: [
|
||||
temp_f: fragment("EXCLUDED.temp_f"),
|
||||
dewpoint_f: fragment("EXCLUDED.dewpoint_f"),
|
||||
relative_humidity: fragment("EXCLUDED.relative_humidity"),
|
||||
wind_speed_kts: fragment("EXCLUDED.wind_speed_kts"),
|
||||
wind_direction_deg: fragment("EXCLUDED.wind_direction_deg"),
|
||||
sea_level_pressure_mb: fragment("EXCLUDED.sea_level_pressure_mb"),
|
||||
altimeter_setting: fragment("EXCLUDED.altimeter_setting"),
|
||||
sky_condition: fragment("EXCLUDED.sky_condition"),
|
||||
precip_1h_in: fragment("EXCLUDED.precip_1h_in"),
|
||||
wx_codes: fragment("EXCLUDED.wx_codes"),
|
||||
updated_at: fragment("EXCLUDED.updated_at")
|
||||
]
|
||||
],
|
||||
where:
|
||||
s.temp_f != fragment("EXCLUDED.temp_f") or
|
||||
s.dewpoint_f != fragment("EXCLUDED.dewpoint_f") or
|
||||
s.relative_humidity != fragment("EXCLUDED.relative_humidity") or
|
||||
s.wind_speed_kts != fragment("EXCLUDED.wind_speed_kts") or
|
||||
s.sea_level_pressure_mb != fragment("EXCLUDED.sea_level_pressure_mb")
|
||||
),
|
||||
conflict_target: [:station_id, :observed_at]
|
||||
)
|
||||
end
|
||||
end
|
||||
|
||||
defp row_has_observed_at?(%{observed_at: %DateTime{}}), do: true
|
||||
defp row_has_observed_at?(%{"observed_at" => %DateTime{}}), do: true
|
||||
defp row_has_observed_at?(_), do: false
|
||||
|
||||
defp surface_observation_entry(row, station_id, now) do
|
||||
%{
|
||||
id: Ecto.UUID.generate(),
|
||||
station_id: station_id,
|
||||
observed_at: fetch_row(row, :observed_at),
|
||||
temp_f: fetch_row(row, :temp_f),
|
||||
dewpoint_f: fetch_row(row, :dewpoint_f),
|
||||
relative_humidity: fetch_row(row, :relative_humidity),
|
||||
wind_speed_kts: fetch_row(row, :wind_speed_kts),
|
||||
wind_direction_deg: fetch_row(row, :wind_direction_deg),
|
||||
sea_level_pressure_mb: fetch_row(row, :sea_level_pressure_mb),
|
||||
altimeter_setting: fetch_row(row, :altimeter_setting),
|
||||
sky_condition: fetch_row(row, :sky_condition),
|
||||
precip_1h_in: fetch_row(row, :precip_1h_in),
|
||||
wx_codes: fetch_row(row, :wx_codes),
|
||||
inserted_at: now,
|
||||
updated_at: now
|
||||
}
|
||||
end
|
||||
|
||||
defp fetch_row(row, key) when is_atom(key) do
|
||||
case Map.fetch(row, key) do
|
||||
{:ok, value} -> value
|
||||
:error -> Map.get(row, Atom.to_string(key))
|
||||
end
|
||||
end
|
||||
|
||||
@spec upsert_sounding(Station.t(), map()) :: {:ok, Sounding.t()} | {:error, Ecto.Changeset.t()}
|
||||
def upsert_sounding(%Station{} = station, attrs) do
|
||||
attrs = Map.put(attrs, :station_id, station.id)
|
||||
|
|
|
|||
|
|
@ -88,11 +88,11 @@ defmodule Microwaveprop.Workers.WeatherFetchWorker do
|
|||
defp ingest_asos_rows(station, station_id, station_code, rows, start_dt, end_dt) do
|
||||
count = Enum.count(rows, & &1.observed_at)
|
||||
|
||||
Enum.each(rows, fn row ->
|
||||
if row.observed_at, do: Weather.upsert_surface_observation(station, row)
|
||||
end)
|
||||
|
||||
if count == 0, do: store_asos_stub(station, start_dt, end_dt)
|
||||
if count > 0 do
|
||||
Weather.upsert_surface_observations(station, rows)
|
||||
else
|
||||
store_asos_stub(station, start_dt, end_dt)
|
||||
end
|
||||
|
||||
Logger.info("WeatherFetch ASOS: #{station_code} ingested #{count} observations")
|
||||
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ defmodule Microwaveprop.WeatherTest do
|
|||
alias Microwaveprop.Weather.HrrrNativeProfile
|
||||
alias Microwaveprop.Weather.HrrrProfile
|
||||
alias Microwaveprop.Weather.IemreObservation
|
||||
alias Microwaveprop.Weather.SurfaceObservation
|
||||
|
||||
@station_attrs %{
|
||||
station_code: "KDFW",
|
||||
|
|
@ -76,6 +77,74 @@ defmodule Microwaveprop.WeatherTest do
|
|||
end
|
||||
end
|
||||
|
||||
describe "upsert_surface_observations/2" do
|
||||
test "bulk inserts all rows in a single call" do
|
||||
{:ok, station} = Weather.find_or_create_station(@station_attrs)
|
||||
|
||||
rows =
|
||||
for i <- 0..99 do
|
||||
%{
|
||||
observed_at: DateTime.add(~U[2026-03-28 00:00:00Z], i * 300, :second),
|
||||
temp_f: 40.0 + i / 10,
|
||||
dewpoint_f: 30.0 + i / 10,
|
||||
sky_condition: "CLR"
|
||||
}
|
||||
end
|
||||
|
||||
assert {count, _} = Weather.upsert_surface_observations(station, rows)
|
||||
assert count == 100
|
||||
assert Repo.aggregate(SurfaceObservation, :count) == 100
|
||||
end
|
||||
|
||||
test "overlapping observed_at updates instead of duplicating" do
|
||||
{:ok, station} = Weather.find_or_create_station(@station_attrs)
|
||||
|
||||
rows1 =
|
||||
for i <- 0..9 do
|
||||
%{
|
||||
observed_at: DateTime.add(~U[2026-03-28 00:00:00Z], i * 300, :second),
|
||||
temp_f: 40.0,
|
||||
dewpoint_f: 30.0
|
||||
}
|
||||
end
|
||||
|
||||
{10, _} = Weather.upsert_surface_observations(station, rows1)
|
||||
|
||||
rows2 =
|
||||
for i <- 0..9 do
|
||||
%{
|
||||
observed_at: DateTime.add(~U[2026-03-28 00:00:00Z], i * 300, :second),
|
||||
temp_f: 80.0,
|
||||
dewpoint_f: 60.0
|
||||
}
|
||||
end
|
||||
|
||||
{_, _} = Weather.upsert_surface_observations(station, rows2)
|
||||
|
||||
assert Repo.aggregate(SurfaceObservation, :count) == 10
|
||||
obs = Repo.all(SurfaceObservation)
|
||||
assert Enum.all?(obs, &(&1.temp_f == 80.0))
|
||||
end
|
||||
|
||||
test "returns {0, _} for empty list without hitting the DB" do
|
||||
{:ok, station} = Weather.find_or_create_station(@station_attrs)
|
||||
assert {0, _} = Weather.upsert_surface_observations(station, [])
|
||||
end
|
||||
|
||||
test "skips rows without observed_at" do
|
||||
{:ok, station} = Weather.find_or_create_station(@station_attrs)
|
||||
|
||||
rows = [
|
||||
%{observed_at: ~U[2026-03-28 18:00:00Z], temp_f: 70.0},
|
||||
%{observed_at: nil, temp_f: 99.0},
|
||||
%{temp_f: 99.0}
|
||||
]
|
||||
|
||||
assert {1, _} = Weather.upsert_surface_observations(station, rows)
|
||||
assert Repo.aggregate(SurfaceObservation, :count) == 1
|
||||
end
|
||||
end
|
||||
|
||||
describe "upsert_sounding/2" do
|
||||
@sample_profile [
|
||||
%{"pres" => 1013.0, "hght" => 171, "tmpc" => 25.0, "dwpc" => 15.0, "drct" => 180, "sknt" => 10},
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue