name: fermihdi-hd-ingest description: Configure an ongoing ingestion pipeline into HD through any supported source — Kafka, RabbitMQ, MQTT, a watched file directory, or an S3-compatible object store — including credentials, formats, batching behavior, and health monitoring. For a one-time bulk load or database migration instead of a standing pipeline, use fermihdi-hd-migrate. Needs a dataset schema first — see fermihdi-hd-schema-build.

Set up an HD ingestion pipeline

This is for standing up a source that keeps feeding HD, not a one-time load — a live Kafka topic, an MQTT feed, a directory or bucket that keeps receiving new files. For moving an existing body of data in once, use fermihdi-hd-migrate; the mechanics below apply there too, but that skill frames it as a bounded job with a start and an end.

Needs a dataset schema first. See fermihdi-hd-schema-build if none exists yet.

Step 1: pick a source

The ingester supports five, set via ingression_type:

If the source needs authentication the ingester can't yet express (a secured Kafka/MQTT broker), say so plainly rather than improvising a config field that doesn't exist — that's a real gap, not a missing step in the skill.

Step 2: pick a format (file and S3 sources only — Kafka/RabbitMQ/MQTT carry pre-formatted records)

CSV, JSON, JSONL/NDJSON, Parquet, ORC, Avro, Protobuf, XML — eight formats total. Parquet/ORC/ Avro carry their own typed schema; CSV/JSON/JSONL don't, so field matching (Step 4) does the work of lining up columns. Protobuf needs formats.protobuf_descriptor_file (a protoc --descriptor_set_out --include_imports output) and formats.protobuf_message_type. XML needs formats.xml_record_element only when records aren't the root's direct children.

Step 3: batching — how records become a batch the storage cluster will accept

A closed batch is never reopened, so getting this right matters more than any other config choice here. The dataset schema's batch_id_field decides the shape:

• A rounded time field (round_batch_id_field: true + time_index_interval/_unit) — the ingester sorts records and computes the batch id itself, automatically, before sending. This is the only form with automatic ordering.
• A raw field value (not rounded) — the batch id is the field's own value, verbatim. No sort happens — the source must already deliver records grouped by that field, or they'll scatter across batches and late arrivals for an already-closed one are refused.
• No batch_id_field at all — the ingester's own clock, rounded by batch_interval_seconds.

batch_close_timeout_mode controls when an open batch closes: dynamic (default — closes after the longest span seen so far), manual (a fixed batch_close_timeout_sec, default 300s), or off (only a later batch closes it — use with care on a source that might stall).

Step 4: field matching — the source doesn't need to match the schema's naming convention

Lookup tries an exact field-name match first, then a normalized fallback (lowercase, strip _/-/space) — so a Kafka payload in camelCase, a CSV header in Title Case, or a Parquet column in snake_case all match a snake_case schema without a producer-side rename. Two source keys that could both match one schema field with neither an exact hit are skipped rather than guessed — check ingester logs for unmapped-field warnings if data seems to be silently missing a column.

Step 5: production schema registration

A real deployment registers its schema through DMS-Core's management API (POST /v1/dms/schema/update, instance_id when more than one deployment exists) — not by hand- editing lab-only provisioning files. Changing record_len on a fleet that's already running is refused unless teardown_acknowledged: true is set, and setting it destroys stored data. Never set that flag without the user's explicit, informed confirmation.

Step 6: monitor it

Health and metrics on port 8081 (/healthz, /metrics). Records-in, batches-closed, and the storage cluster's ack lag are the signals that the pipeline is actually keeping up, not just that the process hasn't crashed. A stalled batch that never closes (source went quiet, batch_close_ timeout_mode: off) is invisible in a simple "is it running" check — watch batch-close events specifically if the source can go idle.

Sizing and jumbo frames

The widest record a dataset may declare is 8433 bytes; a standard frame carries 1133 B of records, jumbo (FERMIHDI_JUMBO=1, instance-wide — SCC, every SN, and the ingester, never just one) 8433 B. If sizing the host itself rather than just the pipeline, use fermihdi-hd-plan for the resource math.

What's actually proven vs. still scaffolding

File-source ingestion is confirmed working end-to-end at real scale, including the batching and field-matching behavior above. S3 is wired up but not load-tested against a real target. Kafka is the documented default and code-complete; RabbitMQ and MQTT exist in code and are documented but weren't recently exercised. None of the three streaming sources (Kafka/RabbitMQ/MQTT) have broker- side authentication wired up — flag that gap rather than assume it's handled.

Troubleshooting

A hung or slow pipeline, records that silently don't arrive, or "already closed" batch errors — see fermihdi-hd-troubleshoot. For a bounded one-time load instead of a standing pipeline, use fermihdi-hd-migrate.