perf: concurrent ASOS fetches + bulk upsert in commercial poll worker
- Replace sequential Enum.each with Task.async_stream for station I/O - Replace per-row upsert_surface_observation with batch upsert_surface_observations
This commit is contained in:
parent
93b8f881e2
commit
fc51f8afe8
1 changed files with 8 additions and 4 deletions
|
|
@ -67,7 +67,12 @@ defmodule Microwaveprop.Commercial.PollWorker do
|
|||
now = DateTime.utc_now()
|
||||
start_dt = DateTime.add(now, -3600, :second)
|
||||
|
||||
Enum.each(stations, &fetch_station_weather(&1, start_dt, now))
|
||||
stations
|
||||
|> Task.async_stream(&fetch_station_weather(&1, start_dt, now),
|
||||
timeout: :infinity,
|
||||
max_concurrency: System.schedulers_online()
|
||||
)
|
||||
|> Stream.run()
|
||||
end
|
||||
|
||||
defp fetch_station_weather(station_code, start_dt, now) do
|
||||
|
|
@ -90,9 +95,8 @@ defmodule Microwaveprop.Commercial.PollWorker do
|
|||
end
|
||||
|
||||
defp ingest_asos(station, station_code, rows) do
|
||||
rows
|
||||
|> Enum.filter(& &1.observed_at)
|
||||
|> Enum.each(&Weather.upsert_surface_observation(station, &1))
|
||||
filtered = Enum.filter(rows, & &1.observed_at)
|
||||
Weather.upsert_surface_observations(station, filtered)
|
||||
|
||||
Logger.info("Fetched #{length(rows)} ASOS observations for #{station_code}")
|
||||
end
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue