Skip to main content

InfluxDB Sink Node

The influxdbsink node writes records to InfluxDB using the v2 write API. Each record is mapped to an InfluxDB point: a measurement name, a set of tags, a set of typed fields, and a timestamp. The mapping is declared with top-level record field names, so no expression language is involved.

Fields not explicitly assigned as tags, timestamp, or measurement source are written as point fields by default, which makes the common case a two-line configuration.

Key Features​

  • Declarative mapping: name the record fields that become tags, values, and the timestamp — everything else follows a sensible default
  • Dynamic measurement routing: take the measurement name from a record field to fan one stream out across many measurements
  • Type preservation: numbers, booleans, and strings are written as their InfluxDB-native types; nested objects and arrays are written as JSON strings
  • Flexible timestamps: use a record field (numeric epoch or ISO-8601 string) or fall back to write time
  • Configurable precision: nanoseconds through seconds
  • Per-record error isolation: a record that cannot be mapped is reported as a single error and the rest of the batch still writes
  • Optional authentication: connect with an API token, or without one for a local/unsecured InfluxDB

Configuration​

FieldTypeRequiredDefaultDescription
urlStringYes—InfluxDB base URL, e.g. https://influx:8086
orgStringYes—InfluxDB organization
bucketStringYes—Target bucket
credentialIdStringNo—ID of an API key credential in jobContext.otherProperties holding the InfluxDB API token. Omit to connect without authentication
measurementStringConditional—Literal measurement name. Exactly one of measurement / measurementField is required
measurementFieldStringConditional—Record field whose value is used as the measurement name. Exactly one of measurement / measurementField is required
tagFieldsList<String>No[]Record fields written as InfluxDB tags. Values are coerced to strings
fieldFieldsList<String>No—Record fields written as InfluxDB fields. When empty, all remaining fields are written (see below)
timestampFieldStringNo—Record field holding the point timestamp. Omit to use write time
precisionStringNoMSTimestamp precision: NS, US, MS, or S
batchSizeintNo1000Maximum records per write batch

All selectors are top-level record field names, not path expressions.

Mapping Records to Points​

Measurement​

Set exactly one of the two — supplying both, or neither, fails validation.

measurement pins a literal name for every point:

config:
  measurement: "cpu"

measurementField reads the name off each record, so one pipeline can write to many measurements:

config:
  measurementField: "metric_name"

With dynamic routing, a record missing that field, or holding a blank value, is reported as an error.

Tags​

tagFields names the record fields written as tags. Tag values are always coerced to strings. A field listed in tagFields but absent from a record is simply skipped — tags are optional per point.

Tags are indexed, so use them for low-cardinality dimensions you filter or group by (host, region, device_id) and avoid high-cardinality values like request ids.

Fields​

fieldFields names the record fields written as point fields. Leaving it empty writes every remaining field — that is, every record key that is not used as a tag, as timestampField, or as measurementField. This default keeps the config short when a record is already shaped like a metric.

Values keep their type:

Record valueWritten as
Integer numberInfluxDB integer
Non-integer numberInfluxDB float
BooleanInfluxDB boolean
StringInfluxDB string
Nested object or arrayJSON string

InfluxDB requires at least one field per point. A record that resolves to zero fields is reported as an error.

Timestamp​

Without timestampField, points are stamped with the pipeline's write time at the configured precision.

With timestampField set, the record must carry it:

Value typeInterpretation
NumberEpoch value already expressed in precision — e.g. with precision: MS, the number is read as epoch milliseconds
StringISO-8601 instant, e.g. 2026-07-23T14:02:11Z
Missing, or another typeRecord is reported as an error

Because a numeric timestamp is interpreted in the configured precision, precision and the units of your timestamp field must agree.

How It Works​

  1. Each record is mapped to an InfluxDB point. A mapping problem — unresolvable measurement, zero fields, or a missing/unparseable timestamp — errors that single record and the batch continues.
  2. Points accumulate up to batchSize, then the batch is written with the synchronous v2 write API.
  3. A write is all-or-nothing: a failure propagates so the sink treats the entire batch as failed. Store-and-forward, if enrolled, can buffer and retry it.

DAG Example​

jobContext:
  otherProperties:
    influx-token:
      key: "my-influxdb-api-token"
  metricTags: {}
  dlqConfig:

dag:
  - id: "source"
    commandName: "stdin"
    config:
      encodingType: "JSON_OBJECT_LINE"
    outputs:
      - "sink"

  - id: "sink"
    commandName: "influxdbsink"
    config:
      url: "https://influx:8086"
      org: "fleak"
      bucket: "metrics"
      credentialId: "influx-token"
      measurement: "host_metrics"
      tagFields:
        - "host"
        - "region"
      timestampField: "ts"
      precision: "MS"
      batchSize: 1000

Given this input record:

{ "host": "web-01", "region": "us-east-1", "cpu": 0.82, "mem_used": 6291456, "ts": 1753280531000 }

host and region become tags, ts becomes the point time, and cpu and mem_used are written as fields — because fieldFields was left empty, they are picked up as the remaining fields. In line protocol:

host_metrics,host=web-01,region=us-east-1 cpu=0.82,mem_used=6291456i 1753280531000