How to Monitor Explorer with Vigilmon

Explorer brings a familiar DataFrame API to Elixir, powered by the high-performance Polars library under the hood. Whether you're building ETL pipelines, real-time analytics, or ML feature engineering, Explorer lets you process large datasets efficiently in Elixir.

Production data pipelines have their own failure modes: a data source that goes stale, a processing job that silently errors on bad rows, or memory pressure from unexpectedly large DataFrames. Standard uptime monitoring won't catch these — you need job heartbeats, data freshness checks, and processing-specific health signals.

This tutorial wires up production monitoring for Explorer-based data pipelines:

• Health endpoints that verify data availability
• HTTP monitoring with Vigilmon
• Heartbeat monitors for scheduled data jobs
• Data freshness and row count alerting
• Memory usage monitoring for large DataFrame operations

Why Monitor Explorer Pipelines?

| Failure mode | What breaks | |---|---| | Data source unavailable | Explorer reads empty/stale files or fails to connect | | Processing job crash | Downstream analytics use outdated data silently | | Memory pressure | Large DataFrames cause OOM kills on the BEAM node | | Bad input data | Unexpected schema breaks transformations | | Slow joins/aggregations | Pipeline falls behind real-time SLAs |

Explorer itself is robust, but the data it processes and the jobs that invoke it need external monitoring.

Step 1: Add a Data Health Endpoint

Add an HTTP health check that verifies your data sources are reachable and recent:

# lib/my_app_web/plugs/data_health.ex
defmodule MyAppWeb.Plugs.DataHealth do
  import Plug.Conn
  require Logger

  @max_data_age_seconds 3600  # Alert if data is more than 1 hour old

  def init(opts), do: opts

  def call(%Plug.Conn{request_path: "/health/data"} = conn, _opts) do
    checks = %{
      source_file: check_source_file(),
      last_processed: check_last_processed(),
      memory: check_memory()
    }

    status = if all_ok?(checks), do: 200, else: 503

    conn
    |> put_resp_content_type("application/json")
    |> send_resp(status, Jason.encode!(%{
      status: if(status == 200, do: "ok", else: "degraded"),
      checks: checks
    }))
    |> halt()
  end

  def call(conn, _opts), do: conn

  defp check_source_file do
    path = Application.get_env(:my_app, :data_source_path, "/data/input.parquet")
    case File.stat(path) do
      {:ok, %{mtime: mtime}} ->
        age = :calendar.datetime_to_gregorian_seconds(:calendar.now_to_universal_time(:erlang.localtime()))
              - :calendar.datetime_to_gregorian_seconds(mtime)
        if age < @max_data_age_seconds, do: :ok, else: :stale
      {:error, _} -> :missing
    end
  end

  defp check_last_processed do
    case MyApp.DataPipeline.last_run_at() do
      nil -> :never_run
      at ->
        age = DateTime.diff(DateTime.utc_now(), at, :second)
        if age < @max_data_age_seconds, do: :ok, else: :stale
    end
  end

  defp check_memory do
    %{total: total, processes: processes} = :erlang.memory() |> Map.new()
    used_mb = processes / 1_048_576
    if used_mb < 1500, do: :ok, else: :high
  end

  defp all_ok?(checks) do
    Enum.all?(checks, fn {_, v} -> v == :ok end)
  end
end

Register in endpoint.ex:

plug MyAppWeb.Plugs.DataHealth
plug MyAppWeb.Router

Step 2: Build a Monitored Explorer Pipeline

Wrap your Explorer operations with monitoring instrumentation:

# lib/my_app/data_pipeline.ex
defmodule MyApp.DataPipeline do
  use GenServer
  require Logger
  alias Explorer.DataFrame, as: DF

  @table "pipeline_state"

  def start_link(opts), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__)

  def last_run_at do
    case :persistent_term.get({__MODULE__, :last_run_at}, nil) do
      nil -> nil
      ts -> DateTime.from_unix!(ts)
    end
  end

  def init(_opts) do
    schedule_run()
    {:ok, %{run_count: 0, error_count: 0}}
  end

  def handle_info(:run, state) do
    start_time = System.monotonic_time(:millisecond)
    result = run_pipeline()
    duration_ms = System.monotonic_time(:millisecond) - start_time

    new_state = case result do
      {:ok, row_count} ->
        Logger.info("Pipeline completed: #{row_count} rows in #{duration_ms}ms")
        :persistent_term.put({__MODULE__, :last_run_at}, System.os_time(:second))
        :telemetry.execute([:my_app, :pipeline, :success], %{duration_ms: duration_ms, rows: row_count}, %{})
        %{state | run_count: state.run_count + 1}

      {:error, reason} ->
        Logger.error("Pipeline failed: #{inspect(reason)}")
        :telemetry.execute([:my_app, :pipeline, :error], %{duration_ms: duration_ms}, %{reason: reason})
        %{state | error_count: state.error_count + 1}
    end

    schedule_run()
    {:noreply, new_state}
  end

  defp run_pipeline do
    source = Application.get_env(:my_app, :data_source_path)

    try do
      df =
        DF.from_parquet!(source)
        |> DF.filter(col("value") > 0)
        |> DF.mutate(
          normalized: col("value") / mean(col("value")),
          processed_at: ^DateTime.utc_now() |> DateTime.to_string()
        )
        |> DF.group_by(["category"])
        |> DF.summarise(
          count: count(col("id")),
          avg_value: mean(col("value"))
        )

      output_path = Application.get_env(:my_app, :output_path)
      DF.to_parquet!(df, output_path)
      {:ok, DF.n_rows(df)}
    rescue
      e -> {:error, Exception.message(e)}
    end
  end

  defp schedule_run, do: Process.send_after(self(), :run, :timer.minutes(5))
end

Step 3: Add Vigilmon HTTP Monitoring

1. Log in to vigilmon.online and click Add Monitor.
2. Set Type to HTTP / HTTPS.
3. Enter https://yourapp.com/health/data.
4. Set Check interval to 2 minutes.
5. Set Expected HTTP status to 200.
6. Under Advanced, add Response body contains: "status":"ok".
7. Click Save.

Step 4: Add a Pipeline Heartbeat

HTTP monitors tell you if the health endpoint is responding. A heartbeat tells you if the pipeline is actually completing successful runs. Configure a Vigilmon heartbeat that only fires after each successful pipeline execution:

# lib/my_app/pipeline_heartbeat.ex
defmodule MyApp.PipelineHeartbeat do
  @moduledoc """
  Pings the Vigilmon heartbeat URL after each successful pipeline run.
  If the pipeline stalls or errors repeatedly, Vigilmon alerts.
  """

  @vigilmon_url "https://vigilmon.online/api/heartbeat/YOUR_HEARTBEAT_ID"

  def ping do
    case HTTPoison.get(@vigilmon_url, [], recv_timeout: 5000) do
      {:ok, %{status_code: 200}} ->
        :ok
      {:ok, %{status_code: code}} ->
        Logger.warning("Vigilmon returned unexpected status: #{code}")
        :ok
      {:error, reason} ->
        Logger.warning("Vigilmon heartbeat failed: #{inspect(reason)}")
        :error
    end
  end
end

Call MyApp.PipelineHeartbeat.ping() at the end of a successful pipeline run:

defp run_pipeline do
  # ... your Explorer transformations ...
  case result do
    {:ok, row_count} ->
      MyApp.PipelineHeartbeat.ping()
      {:ok, row_count}
    error -> error
  end
end

Create the heartbeat monitor in Vigilmon:

1. Click Add Monitor → Cron / Heartbeat.
2. Set the expected interval to match your pipeline schedule (e.g., 5 minutes).
3. Set grace period to 10 minutes (allows for slow runs before alerting).
4. Copy the heartbeat URL into @vigilmon_url.

Step 5: Monitor Data Freshness and Row Counts

Validate that processed data meets quality thresholds before marking a run successful:

# lib/my_app/data_validator.ex
defmodule MyApp.DataValidator do
  alias Explorer.DataFrame, as: DF
  require Logger

  @min_rows 100
  @max_null_ratio 0.05  # Alert if > 5% nulls in key columns

  def validate(df) do
    validations = [
      check_row_count(df),
      check_null_ratio(df, "value"),
      check_schema(df)
    ]

    case Enum.find(validations, fn v -> v != :ok end) do
      nil -> {:ok, DF.n_rows(df)}
      error -> {:error, error}
    end
  end

  defp check_row_count(df) do
    count = DF.n_rows(df)
    if count >= @min_rows do
      :ok
    else
      "Row count too low: #{count} (minimum: #{@min_rows})"
    end
  end

  defp check_null_ratio(df, column) do
    total = DF.n_rows(df)
    return if total == 0, do: :ok

    null_count =
      df
      |> DF.filter(is_nil(col(^column)))
      |> DF.n_rows()

    ratio = null_count / total
    if ratio <= @max_null_ratio do
      :ok
    else
      "Null ratio too high in #{column}: #{Float.round(ratio * 100, 1)}%"
    end
  end

  defp check_schema(df) do
    required_columns = ["id", "category", "value", "timestamp"]
    actual_columns = DF.names(df)
    missing = required_columns -- actual_columns
    if missing == [], do: :ok, else: "Missing columns: #{inspect(missing)}"
  end
end

Use the validator in your pipeline:

df = DF.from_parquet!(source) |> transform()

case MyApp.DataValidator.validate(df) do
  {:ok, row_count} ->
    DF.to_parquet!(df, output_path)
    MyApp.PipelineHeartbeat.ping()
    {:ok, row_count}
  {:error, reason} ->
    Logger.error("Data validation failed: #{reason}")
    {:error, reason}
end

Step 6: Monitor Memory Usage for Large DataFrames

Explorer loads DataFrames into native memory via Polars. Large datasets can exhaust BEAM memory. Add a memory check:

# lib/my_app/memory_monitor.ex
defmodule MyApp.MemoryMonitor do
  use GenServer
  require Logger

  @check_interval :timer.seconds(30)
  @warn_threshold_mb 1_024   # 1 GB
  @alert_threshold_mb 1_536  # 1.5 GB

  def start_link(_), do: GenServer.start_link(__MODULE__, nil, name: __MODULE__)

  def init(_) do
    schedule()
    {:ok, nil}
  end

  def handle_info(:check, state) do
    memory = :erlang.memory()
    total_mb = Keyword.get(memory, :total, 0) / 1_048_576

    cond do
      total_mb > @alert_threshold_mb ->
        Logger.error("Memory critical: #{Float.round(total_mb, 0)} MB — risk of OOM")
        :telemetry.execute([:my_app, :memory, :critical], %{mb: total_mb}, %{})

      total_mb > @warn_threshold_mb ->
        Logger.warning("Memory elevated: #{Float.round(total_mb, 0)} MB")
        :telemetry.execute([:my_app, :memory, :warning], %{mb: total_mb}, %{})

      true -> :ok
    end

    schedule()
    {:noreply, state}
  end

  defp schedule, do: Process.send_after(self(), :check, @check_interval)
end

Add MyApp.MemoryMonitor to your supervision tree.

Step 7: Set Up Alerts

Configure Vigilmon to alert your team when Explorer pipelines degrade:

1. Go to Alert Channels → Add Channel.
2. Add Slack and/or Email.
3. Set your data health monitor to alert after 2 consecutive failures.
4. Set your heartbeat monitor to alert when the ping is 10 minutes overdue (adjust based on your pipeline schedule).

For data teams, also configure a Status Page to communicate processing status:

1. Go to Status Pages → Create Page.
2. Add both monitors (HTTP health + heartbeat).
3. Rename them to user-facing labels like "Data Processing" and "Analytics Pipeline".
4. Share the URL with internal stakeholders.

Key Metrics to Watch

| Metric | Vigilmon feature | Alert threshold | |---|---|---| | Data source availability | HTTP monitor on /health/data | Any 5xx | | Data freshness | HTTP monitor body check | "last_processed":"stale" | | Pipeline completion | Heartbeat monitor | Missing for > 2× schedule interval | | Row count validation | Embedded in heartbeat gate | < 100 rows after transform | | Memory pressure | Custom GenServer + logging | > 1.5 GB total | | SSL certificate | Cert expiry monitor | < 14 days remaining |

Conclusion

Explorer makes data manipulation expressive and fast in Elixir, but production data pipelines need observability beyond what the library itself provides. With a data freshness health endpoint, a completion heartbeat, row-count validation, and memory monitoring, Vigilmon gives you full visibility into whether your Explorer pipelines are processing fresh, valid data on schedule.

Start monitoring your data pipelines for free at vigilmon.online.