Fleet telematics to Iceberg

Location pings from two fleets into one PicoMQ stream per fleet, then one Iceberg table.

Source: examples/connectors/fleet-telematics-iceberg.

Run

cd examples/connectors/fleet-telematics-iceberg
docker compose up -d --build
./seed.sh
./verify.sh

Service	Host
PicoMQ	http://localhost:9090
Connectors runtime	http://localhost:8081
Iceberg REST catalog	http://localhost:8181
RustFS	http://localhost:9001
DuckDB	docker compose exec duckdb duckdb -init /opt/duckdb.sql

--build compiles the Iceberg sink image. It is not in the default connectors image.

Sink

iceberg-sink.toml:

[[topics]]
pattern = "fleets\\..*"

[plugin_config]
tables = ["analytics.locations"]
catalog_type = "rest"
uri = "http://iceberg-rest:8181"
warehouse = "s3://lake/warehouse/"
store_url = "http://rustfs:9000"
store_path_style_access = true

The table is not created by the sink. createtable in compose.yml posts sql/namespace.json and sql/table.json to the catalog first.

tables	Result
analytics.locations	One table, fleet_id as a column
analytics.locations_{topic_segment[-1]}	analytics.locations_a, analytics.locations_b

Verify

	PicoMQ	Iceberg (DuckDB)
Fleets	pico ls --prefix /fleets.	SELECT fleet_id, count(*) FROM lake.analytics.locations GROUP BY 1
Fleet A	pico read /fleets.a	SELECT * FROM lake.analytics.locations WHERE fleet_id = 'a'

Resume

docker compose kill connectors
printf '%s\n' '{"id":"a-resume","fleet_id":"a","vehicle_id":"van-1","lat":51.51,"lon":-0.12,"ts":"2026-09-05T21:00:00Z"}' \
  | docker compose exec -T pico pico append /fleets.a --batch 1
docker compose start connectors
./verify.sh