MLOps with Ray and Python on AIS Vessel Data

Most MLOps examples use tiny tabular datasets that fit comfortably into a notebook and never have to deal with time, geography, or noisy telemetry. AIS data is the opposite. It is messy, high-volume, and operational by nature.

That makes it a good fit for Ray.

In this post, we will use the tayljordan/ais dataset, a public JSON sample of AIS vessel records derived from AISHub and reduced to 10,000 vessels. The repository describes fields such as MMSI, TSTAMP, LATITUDE, LONGITUDE, COG, SOG, HEADING, NAVSTAT, TYPE, DRAUGHT, DEST, and ETA.

The goal is not to build a perfect maritime model. The goal is to show how Ray helps you turn raw AIS records into a repeatable Python MLOps workflow for:

• distributed ingestion
• parallel feature engineering
• scalable training
• reproducible batch inference
• production-friendly orchestration

Why Ray for MLOps?

Ray is useful when your ML system stops being “one script, one machine.”

AIS workloads hit that point quickly:

• many vessels
• many timestamps per vessel
• trajectory-aware features
• geospatial and temporal preprocessing
• retraining and scoring jobs that benefit from horizontal scale

Instead of stitching together separate systems for parallel Python, data pipelines, training, and serving, Ray gives you one runtime with multiple layers:

• ray.data for distributed ingestion and preprocessing
• ray.train for parallel model training
• ray[tune] for hyperparameter search
• ray serve for model deployment
• Ray Jobs or scheduled workflows for repeatable execution

The AIS Dataset Shape

According to the repository README, the dataset is a curated JSON file containing vessel-level AIS observations sampled from a larger AISHub source. A representative record looks like this:

{
  "MMSI": "257465900",
  "TSTAMP": "2024-11-19 20:28:06 GMT",
  "LATITUDE": "63.43565",
  "LONGITUDE": "10.39761",
  "COG": "137.6",
  "SOG": "0",
  "HEADING": "511",
  "NAVSTAT": "15",
  "IMO": "0",
  "NAME": "D/S HANSTEEN",
  "CALLSIGN": "LDQF",
  "TYPE": "60",
  "A": "20",
  "B": "10",
  "C": "3",
  "D": "2",
  "DRAUGHT": "0",
  "DEST": "",
  "ETA": "00-00 00:00"
}

I can say that the schema is enough to build practical MLOps demos:

• predict vessel type from movement and geometry fields
• classify whether a record looks operationally abnormal
• forecast short-horizon speed buckets
• detect route deviations per vessel

For this walkthrough, we will build a simple classification pipeline that predicts a normalized vessel type label from AIS movement and dimensional features.

Architecture

Here is the high-level flow:

1. Pull the JSON dataset into object storage or a shared filesystem.
2. Use ray.data to load and clean records in parallel.
3. Create features such as numeric speed, heading validity, vessel dimensions, and timestamp features.
4. Split data into train and validation sets.
5. Train a model with Ray on multiple workers.
6. Register model artifacts and metrics.
7. Deploy a batch or online inference endpoint with Ray Serve.

This is what makes it MLOps rather than just ML:

• the pipeline is repeatable
• data cleaning is code, not notebook state
• training is trackable
• inference uses the same transformations
• scaling does not require rewriting the pipeline

Project Layout

Keep the repository boring and predictable:

ray-ais-mlops/
|-- data/
|   |-- raw/
|   |   |-- ais_data.json
|-- src/
|   |-- ingest.py
|   |-- features.py
|   |-- train.py
|   |-- batch_score.py
|   |-- serve.py
|-- models/
|-- reports/
|-- requirements.txt
|-- ray-job.yaml

Install the Stack

pip install "ray[data,train,tune,serve]" pandas pyarrow scikit-learn xgboost mlflow

If you want local experiment tracking too:

pip install mlflow
mlflow ui

Step 1: Load AIS Data with Ray Data

Ray Data is a nice fit here because raw AIS files are often line-oriented, semi-structured, and large enough that pandas alone becomes a bottleneck.

import ray
from ray.data import read_json

ray.init()

ds = read_json("data/raw/ais_data.json")
print(ds.schema())
print(ds.count())

Once loaded, normalize types early:

from datetime import datetime

NUMERIC_COLS = ["LATITUDE", "LONGITUDE", "COG", "SOG", "HEADING", "TYPE", "A", "B", "C", "D", "DRAUGHT"]

def clean_batch(df):
    for col in NUMERIC_COLS:
        if col in df.columns:
            df[col] = df[col].astype("float64", errors="ignore")

    if "TSTAMP" in df.columns:
        df["TSTAMP"] = df["TSTAMP"].astype("string")
        df["event_hour"] = df["TSTAMP"].str[11:13]

    df["beam"] = df.get("C", 0).fillna(0) + df.get("D", 0).fillna(0)
    df["length"] = df.get("A", 0).fillna(0) + df.get("B", 0).fillna(0)
    df["is_moving"] = (df.get("SOG", 0).fillna(0) > 0.5).astype("int8")
    df["heading_missing"] = (df.get("HEADING", 511).fillna(511) >= 511).astype("int8")

    return df

cleaned = ds.map_batches(clean_batch, batch_format="pandas")

A few small details matter a lot with AIS:

• HEADING=511 is commonly used as “not available”
• SOG=0 can mean anchored, berthed, or stale telemetry
• empty DEST and placeholder ETA values are normal
• dimensional fields A/B/C/D are useful because they approximate vessel length and beam

Step 2: Build Features in Parallel

For an initial model, do not overcomplicate things. Start with robust, cheap features:

• latitude
• longitude
• speed over ground
• course over ground
• heading availability
• draught
• length
• beam
• hour of day

FEATURES = [
    "LATITUDE",
    "LONGITUDE",
    "SOG",
    "COG",
    "heading_missing",
    "DRAUGHT",
    "length",
    "beam",
    "event_hour",
]

TARGET = "TYPE"

filtered = cleaned.filter(
    lambda row: row.get("TYPE") not in [None, "", 0, "0"]
)

train_ds, valid_ds = filtered.train_test_split(test_size=0.2, seed=42)

On bigger AIS corpora, this is where Ray starts paying off. Feature blocks can run across many partitions without you manually managing multiprocessing pools.

Step 3: Train a Model with Ray

For tabular AIS classification, gradient-boosted trees are a reasonable baseline. XGBoost works well, and Ray lets you scale training when the dataset grows beyond one host.

import os
import mlflow
import pandas as pd
from xgboost import XGBClassifier
from sklearn.metrics import accuracy_score, f1_score

def train_loop(config):
    train_df = train_ds.to_pandas()[FEATURES + [TARGET]].dropna()
    valid_df = valid_ds.to_pandas()[FEATURES + [TARGET]].dropna()

    X_train = train_df[FEATURES]
    y_train = train_df[TARGET].astype(int)
    X_valid = valid_df[FEATURES]
    y_valid = valid_df[TARGET].astype(int)

    model = XGBClassifier(
        n_estimators=config["n_estimators"],
        max_depth=config["max_depth"],
        learning_rate=config["learning_rate"],
        subsample=config["subsample"],
        colsample_bytree=config["colsample_bytree"],
        tree_method="hist",
        eval_metric="mlogloss",
    )

    with mlflow.start_run():
        model.fit(X_train, y_train)
        preds = model.predict(X_valid)

        acc = accuracy_score(y_valid, preds)
        f1 = f1_score(y_valid, preds, average="weighted")

        mlflow.log_params(config)
        mlflow.log_metric("accuracy", acc)
        mlflow.log_metric("f1_weighted", f1)
        mlflow.sklearn.log_model(model, "model")

        print({"accuracy": acc, "f1_weighted": f1})

For local development, that is enough. For cluster execution, wrap it with Ray Tune:

from ray import tune

tuner = tune.Tuner(
    train_loop,
    param_space={
        "n_estimators": tune.choice([100, 200, 300]),
        "max_depth": tune.choice([4, 6, 8]),
        "learning_rate": tune.choice([0.03, 0.05, 0.1]),
        "subsample": tune.choice([0.8, 1.0]),
        "colsample_bytree": tune.choice([0.8, 1.0]),
    },
)

results = tuner.fit()

The benefit is not just speed. It is consistency. Every run uses the same feature pipeline and the same parameterized training entrypoint.

Step 4: Batch Inference with Ray

AIS systems often care more about batch scoring than fancy online demos. Nightly enrichment jobs, anomaly sweeps, and fleet summaries are common.

Ray makes that straightforward:

import mlflow.pyfunc

model = mlflow.pyfunc.load_model("models:/ais-vessel-type/latest")

def score_batch(df):
    features = df[FEATURES].fillna(0)
    df["predicted_type"] = model.predict(features)
    return df

scored = cleaned.map_batches(score_batch, batch_format="pandas")
scored.write_parquet("reports/scored_ais")

This is a good pattern when:

• you need to score millions of records
• one machine is too slow
• you want simple Python without rewriting into Spark jobs

Step 5: Online Serving with Ray Serve

If you later want online inference, reuse the same model artifact:

from ray import serve
import mlflow.pyfunc
import pandas as pd

@serve.deployment(num_replicas=2)
class VesselTypeAPI:
    def __init__(self):
        self.model = mlflow.pyfunc.load_model("models:/ais-vessel-type/latest")

    async def __call__(self, request):
        payload = await request.json()
        frame = pd.DataFrame(payload["inputs"])
        preds = self.model.predict(frame[FEATURES])
        return {"predictions": preds.tolist()}

serve.run(VesselTypeAPI.bind())

That gives you a Python-native deployment path without introducing a separate serving stack too early.

What to Monitor

MLOps starts getting real once you monitor behavior instead of just training metrics.

For AIS, useful checks include:

• feature drift in SOG, COG, DRAUGHT, and vessel geography
• share of missing headings
• growth in unseen or rare TYPE labels
• prediction distribution shifts by port, region, or hour
• latency and throughput for batch and online scoring

Even a simple weekly job that compares current distributions against the training window will catch a lot of operational pain.

Common AIS Data Problems

Ray helps with scale, but it does not fix bad telemetry for you.

Expect to handle:

• duplicated reports
• impossible jumps in latitude and longitude
• stale timestamps
• bad vessel dimensions
• placeholder values like 511 for heading
• sparse destination metadata

That is why feature engineering should live in versioned Python modules, not in ad hoc notebook cells.

Where Ray Fits Best

Ray is a strong choice when:

• your team is already comfortable in Python
• pandas workflows are becoming too slow
• you want one system for preprocessing, training, tuning, and serving
• your data volume is growing but you do not want a heavyweight platform too early

Ray is probably not the first tool to reach for when:

• the data still fits comfortably in one machine
• your bottleneck is labeling quality rather than compute
• you only need a nightly cron job and a single sklearn model

A Practical Next Step

If you want to turn this into a stronger maritime ML system, the next upgrade is not a fancier model. It is better labels and better sequence features.

Single AIS rows are often too thin. One record tells you where a vessel was at one instant. It does not tell you whether the vessel is accelerating strangely, drifting near a restricted zone, or behaving unusually relative to its own recent path.

The practical next step is to move from row-level learning to sequence-aware learning.

That usually means five upgrades:

• aggregate records per vessel over 30-minute windows
• compute rolling speed and course statistics
• add distance-to-port or geofenced regions
• create anomaly labels from impossible motion patterns
• train on trajectory segments instead of single rows

Here is what that looks like in Python.

1. Aggregate Per Vessel Over 30-Minute Windows

Instead of treating every AIS message as an independent sample, group messages by MMSI and resample them into fixed time windows. That gives you one training row per vessel-window instead of one row per transmission.

This is usually a much better unit for:

• vessel behavior classification
• route-stage detection
• anomaly detection
• ETA modeling

import pandas as pd

def build_time_windows(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()
    df["TSTAMP"] = pd.to_datetime(df["TSTAMP"], utc=True, errors="coerce")
    df["SOG"] = pd.to_numeric(df["SOG"], errors="coerce")
    df["COG"] = pd.to_numeric(df["COG"], errors="coerce")
    df["LATITUDE"] = pd.to_numeric(df["LATITUDE"], errors="coerce")
    df["LONGITUDE"] = pd.to_numeric(df["LONGITUDE"], errors="coerce")
    df["DRAUGHT"] = pd.to_numeric(df["DRAUGHT"], errors="coerce")

    df = df.dropna(subset=["MMSI", "TSTAMP", "LATITUDE", "LONGITUDE"])
    df = df.sort_values(["MMSI", "TSTAMP"])

    aggregated = (
        df.set_index("TSTAMP")
        .groupby("MMSI")
        .resample("30min")
        .agg(
            lat_mean=("LATITUDE", "mean"),
            lon_mean=("LONGITUDE", "mean"),
            sog_mean=("SOG", "mean"),
            sog_max=("SOG", "max"),
            cog_mean=("COG", "mean"),
            draught_mean=("DRAUGHT", "mean"),
            msg_count=("LATITUDE", "count"),
            vessel_type=("TYPE", "first"),
        )
        .reset_index()
    )

    return aggregated.dropna(subset=["lat_mean", "lon_mean"])

If you run this through Ray, the same logic can be applied partition by partition:

windowed = cleaned.map_batches(build_time_windows, batch_format="pandas")

One warning: for very large AIS corpora, you usually want all rows for the same MMSI to land in the same partition before resampling. That is where repartitioning and grouped processing become important.

2. Compute Rolling Speed and Course Statistics

Behavior is often defined by change, not by state.

A cargo vessel cruising steadily at 14 knots is boring. A vessel oscillating between 0, 18, and 2 knots over a short interval is interesting. A course that swings 120 degrees in a narrow window may be a maneuver, a tug operation, a reporting glitch, or something worth investigating.

So compute rolling statistics per vessel:

import numpy as np

def add_rolling_features(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()
    df = df.sort_values(["MMSI", "TSTAMP"])

    grouped = df.groupby("MMSI", group_keys=False)

    df["sog_roll_mean_6"] = grouped["SOG"].transform(
        lambda s: s.rolling(6, min_periods=1).mean()
    )
    df["sog_roll_std_6"] = grouped["SOG"].transform(
        lambda s: s.rolling(6, min_periods=1).std().fillna(0)
    )
    df["sog_delta"] = grouped["SOG"].diff().fillna(0)

    df["cog_delta"] = grouped["COG"].diff().fillna(0)
    df["cog_delta"] = ((df["cog_delta"] + 180) % 360) - 180
    df["cog_abs_delta"] = df["cog_delta"].abs()

    df["lat_delta"] = grouped["LATITUDE"].diff().fillna(0)
    df["lon_delta"] = grouped["LONGITUDE"].diff().fillna(0)

    return df

This gives you features like:

• average recent speed
• speed volatility
• sudden acceleration or deceleration
• abrupt course changes
• tiny or large spatial movement between reports

Those are much more useful for maritime behavior modeling than raw SOG and COG alone.

3. Add Distance-to-Port or Geofenced Regions

Raw latitude and longitude are fine, but domain features are better. One of the strongest domain signals in AIS is location relative to meaningful maritime areas:

• port approach zones
• anchorage areas
• traffic separation schemes
• restricted zones
• offshore platforms

You can start simple with a hand-built geofence table.

from math import radians, sin, cos, sqrt, atan2

PORTS = [
    {"name": "Rotterdam", "lat": 51.9244, "lon": 4.4777},
    {"name": "Hamburg", "lat": 53.5461, "lon": 9.9661},
    {"name": "Singapore", "lat": 1.2644, "lon": 103.8223},
]

def haversine_km(lat1, lon1, lat2, lon2):
    r = 6371.0
    dlat = radians(lat2 - lat1)
    dlon = radians(lon2 - lon1)
    a = (
        sin(dlat / 2) ** 2
        + cos(radians(lat1)) * cos(radians(lat2)) * sin(dlon / 2) ** 2
    )
    return 2 * r * atan2(sqrt(a), sqrt(1 - a))

def nearest_port_features(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()
    nearest_names = []
    nearest_distances = []

    for _, row in df.iterrows():
        distances = [
            (
                port["name"],
                haversine_km(row["LATITUDE"], row["LONGITUDE"], port["lat"], port["lon"]),
            )
            for port in PORTS
        ]
        port_name, dist_km = min(distances, key=lambda x: x[1])
        nearest_names.append(port_name)
        nearest_distances.append(dist_km)

    df["nearest_port"] = nearest_names
    df["dist_to_port_km"] = nearest_distances
    df["near_port"] = (df["dist_to_port_km"] < 20).astype("int8")
    return df

For production systems, you usually replace the row loop with a vectorized spatial join or a proper geospatial index, but the feature idea stays the same.

You can also define simple rectangular geofences:

def add_geofence_flags(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()

    df["in_rotterdam_box"] = (
        (df["LATITUDE"].between(51.85, 52.05)) &
        (df["LONGITUDE"].between(4.2, 4.7))
    ).astype("int8")

    df["in_hamburg_box"] = (
        (df["LATITUDE"].between(53.45, 53.65)) &
        (df["LONGITUDE"].between(9.8, 10.2))
    ).astype("int8")

    return df

These kinds of features are often more predictive than raw coordinates because they encode operational context.

4. Create Anomaly Labels from Impossible Motion Patterns

Real AIS anomaly projects usually fail because the labels are weak. If you wait for perfect human labels, you may never ship. A practical compromise is to generate high-confidence heuristic labels first.

For example:

• speed over 45 knots for a vessel that should not do that
• huge geographic jump in a short time
• impossible turn rate
• reporting from far outside the recent route corridor

Here is a simple heuristic labeler:

def label_motion_anomalies(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()
    df["TSTAMP"] = pd.to_datetime(df["TSTAMP"], utc=True, errors="coerce")
    df = df.sort_values(["MMSI", "TSTAMP"])

    grouped = df.groupby("MMSI", group_keys=False)

    df["prev_lat"] = grouped["LATITUDE"].shift(1)
    df["prev_lon"] = grouped["LONGITUDE"].shift(1)
    df["prev_ts"] = grouped["TSTAMP"].shift(1)
    df["time_gap_sec"] = (df["TSTAMP"] - df["prev_ts"]).dt.total_seconds()

    def row_distance(row):
        if pd.isna(row["prev_lat"]) or pd.isna(row["prev_lon"]):
            return np.nan
        return haversine_km(
            row["prev_lat"], row["prev_lon"], row["LATITUDE"], row["LONGITUDE"]
        )

    df["jump_km"] = df.apply(row_distance, axis=1)
    df["jump_km_per_min"] = df["jump_km"] / (df["time_gap_sec"] / 60.0)

    df["anomaly_label"] = (
        (df["SOG"] > 45) |
        (df["jump_km"] > 30) |
        (df["jump_km_per_min"] > 2)
    ).fillna(False).astype("int8")

    return df

This is not your final truth set. It is your bootstrap label set.

A very common production pattern is:

1. start with heuristic anomaly labels
2. train a first-pass anomaly model
3. send high-risk and low-confidence examples for review
4. fold reviewed cases back into the training set

That gives you a path from rules to learned behavior without waiting six months for annotation.

5. Train on Trajectory Segments Instead of Single Rows

Once you have sequence-aware features, the next upgrade is to train on short trajectory segments.

For example, instead of one row per AIS message, build one training sample from the last 12 points of a vessel track. That lets the model learn local motion patterns.

Here is a straightforward segment builder:

def build_segments(df: pd.DataFrame, segment_len: int = 12) -> pd.DataFrame:
    df = df.copy()
    df["TSTAMP"] = pd.to_datetime(df["TSTAMP"], utc=True, errors="coerce")
    df = df.sort_values(["MMSI", "TSTAMP"])

    segment_rows = []

    for mmsi, vessel_df in df.groupby("MMSI"):
        vessel_df = vessel_df.reset_index(drop=True)

        for start in range(0, len(vessel_df) - segment_len + 1):
            segment = vessel_df.iloc[start:start + segment_len]

            row = {
                "MMSI": mmsi,
                "segment_start": segment["TSTAMP"].iloc[0],
                "segment_end": segment["TSTAMP"].iloc[-1],
                "lat_mean": segment["LATITUDE"].mean(),
                "lon_mean": segment["LONGITUDE"].mean(),
                "sog_mean": segment["SOG"].mean(),
                "sog_std": segment["SOG"].std(),
                "sog_max": segment["SOG"].max(),
                "cog_abs_delta_mean": segment["cog_abs_delta"].mean(),
                "draught_mean": segment["DRAUGHT"].mean(),
                "msg_count": len(segment),
                "target_type": segment["TYPE"].mode().iloc[0],
                "target_anomaly": int(segment["anomaly_label"].max()),
            }

            segment_rows.append(row)

    return pd.DataFrame(segment_rows)

This produces one feature row per short trajectory slice. You can use it for:

• vessel type prediction
• anomaly classification
• route family classification
• loitering detection

If you later move to an LSTM, Temporal CNN, or Transformer-style sequence model, this segment-oriented dataset becomes your bridge.

Putting the Feature Pipeline Together

The nice part is that these upgrades compose cleanly:

def full_feature_pipeline(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()
    df["SOG"] = pd.to_numeric(df["SOG"], errors="coerce")
    df["COG"] = pd.to_numeric(df["COG"], errors="coerce")
    df["LATITUDE"] = pd.to_numeric(df["LATITUDE"], errors="coerce")
    df["LONGITUDE"] = pd.to_numeric(df["LONGITUDE"], errors="coerce")
    df["DRAUGHT"] = pd.to_numeric(df["DRAUGHT"], errors="coerce")

    df = add_rolling_features(df)
    df = nearest_port_features(df)
    df = add_geofence_flags(df)
    df = label_motion_anomalies(df)
    return df

featured = cleaned.map_batches(full_feature_pipeline, batch_format="pandas")
segment_ds = featured.map_batches(build_segments, batch_format="pandas")

And the segment dataset can feed the same Ray training setup:

SEGMENT_FEATURES = [
    "lat_mean",
    "lon_mean",
    "sog_mean",
    "sog_std",
    "sog_max",
    "cog_abs_delta_mean",
    "draught_mean",
    "msg_count",
    "dist_to_port_km",
    "near_port",
    "in_rotterdam_box",
    "in_hamburg_box",
]

segment_train, segment_valid = segment_ds.train_test_split(test_size=0.2, seed=42)

That is the main idea: the next improvement is not primarily a bigger model. It is a better representation of vessel behavior.

Ray still helps here, because these transformations are exactly where dataset size and parallelism start to matter. Sequence features, resampling, rolling windows, heuristic labels, and segment generation are all more expensive than simple row-level training, but they are also much closer to the real maritime problems you actually want to solve.

Final Thoughts

The tayljordan/ais repository is a useful starter dataset because it is simple enough to explore quickly but realistic enough to expose the kinds of problems AIS pipelines actually have.

Ray gives you a clean path from “I can process this in Python” to “I can process this repeatedly and at scale.” That is the real MLOps win.

Not more dashboards. Not more architecture diagrams. Just one pipeline that keeps working as the data gets bigger.