> For the complete documentation index, see [llms.txt](https://docs.cybus.io/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.cybus.io/connectors/enterprise-connectors/influxdb.md). # InfluxDB Connect Connectware to InfluxDB, a time series database designed for storing and querying time-stamped data. This integration lets you read from and write to your InfluxDB instance using MQTT topics. InfluxDB organizes data into measurements (similar to tables), where each measurement contains data points. Each point includes a timestamp, fields that hold the data values, and tags that you can use to organize and filter your data. Connectware supports InfluxDB 2 and InfluxDB 3: * The `Influxdb` protocol targets InfluxDB 2 and uses the Flux query language. See [InfluxDB 2 Support](#influxdb-2-support). * The `Influxdb3` protocol targets InfluxDB 3 (Core, Enterprise, and Cloud) and uses SQL or InfluxQL. See [InfluxDB 3 Support](#influxdb-3-support). ## InfluxDB 2 Support Connectware integrates with InfluxDB 2 using the `Influxdb` protocol. * [Connection Properties](/connectors/enterprise-connectors/influxdb/influxdbconnection.md) * [Endpoint Properties](/connectors/enterprise-connectors/influxdb/influxdbendpoint.md) Set the `org` and `bucket` connection properties to select the InfluxDB organization and bucket that Connectware reads from and writes to. ### Reading Data You can read data from InfluxDB in the following ways: * **One-time read**: Define an endpoint with a `read` property to fetch data on demand. * **Continuous subscription**: Define an endpoint with a `subscribe` property to poll data at regular intervals. Both methods require a valid Flux query in the endpoint's `query` property. When you subscribe to data, specify a polling interval. The query runs automatically at that frequency. **Dynamic queries**: Use `@` placeholders in your Flux queries to make them flexible and reusable. {% code lineNumbers="true" %} ``` from(bucket:"@bucket") |> range(start: @startMeasurementTime) |> filter(fn: (r) => r._measurement == "@measurement") ``` {% endcode %} In this example, when you request a read and provide values for `bucket`, `measurement`, and `startMeasurementTime`, Connectware generates the complete Flux query by replacing the `@` placeholders with your values. ### Output Format on Read Connectware publishes query results as JSON to MQTT topics: * **One-time reads**: Results appear on the endpoint's `/res` topic. * **Subscriptions**: Results appear on the endpoint's default topic. You receive the data as a JSON array containing your InfluxDB query results. {% code lineNumbers="true" %} ```json [ { "result": "_result", "table": 0, "_start": "2021-02-14T09:29:24.514083303Z", "_stop": "2021-02-15T09:29:24.514083303Z", "_time": "2021-02-15T09:29:06.059Z", "_value": 19.7, "_field": "value", "_measurement": "temperature" }, { "result": "_result", "table": 0, "_start": "2021-02-14T09:29:24.514083303Z", "_stop": "2021-02-15T09:29:24.514083303Z", "_time": "2021-02-15T09:29:06.059623817Z", "_value": 21.3, "_field": "value", "_measurement": "temperature" } ] ``` {% endcode %} ### Writing Data To write data to InfluxDB, define an endpoint with a `write` property. **Setting the measurement name**: You can set a default `measurement` property in your endpoint configuration. This applies to all data points you send. To override this default, include a `measurement` property in individual data messages. **How to write**: Send an MQTT message to the endpoint's `/set` topic. The following example includes a measurement name in the message: {% code lineNumbers="true" %} ```json { "tags": { "rpm": "8000", "oil_temp": "250" }, "value": 91, "fields": { "engine_number": 1 }, "timestamp": 1719129600000, "measurement": "temperature" } ``` {% endcode %} **What you can include**: * **Tags and fields**: Use both to organize your data. * **Timestamp**: Optional. If you do not provide one, InfluxDB assigns the current time. **Writing multiple points**: Send an array of data points in a single message to write multiple values at once. **Write behavior**: Writes are asynchronous. Connectware buffers data points and writes them to InfluxDB at intervals defined by the `flushInterval` connection property (default: 10 seconds). This follows InfluxDB's recommended client design pattern. ### Output Format on Write After writing data, you receive a confirmation message on the endpoint's `/res` topic. This response contains: * **timestamp**: Unix timestamp (in milliseconds) of the write. * **value**: Set to `true` when the write succeeds. ### Service Commissioning File Example This example demonstrates a complete InfluxDB 2 integration with write endpoints and a subscription for reading data. {% file src="/files/ExZNuOhDZnxiBiRRmJEN" %} {% code title="influxdb-example.yml" lineNumbers="true" %} ```yaml description: | Sample InfluxDB commissioning file metadata: name: Cybus InfluxDB Example provider: cybus homepage: https://cybus.io version: 1.0.0 #------------------------------------------------------------------------------ # Parameters #------------------------------------------------------------------------------ parameters: influxHost: type: string description: 'HTTP address of InfluxDB server' default: 'influxdbhost' influxPort: type: integer description: 'Influx Port' default: 8086 influxScheme: type: 'string' description: 'Either use http or https for the server url' default: 'http' #------------------------------------------------------------------------------ # Resources #------------------------------------------------------------------------------ resources: #---------------------------------------------------------------------------- # Connections #---------------------------------------------------------------------------- influxdbConnection: type: Cybus::Connection properties: protocol: Influxdb connection: host: !ref influxHost token: '-an-influx-db-jwt-token-' port: !ref influxPort bucket: turbine scheme: !ref influxScheme flushInterval: 5000 #---------------------------------------------------------------------------- # Endpoints #---------------------------------------------------------------------------- turbineWrite: type: Cybus::Endpoint properties: protocol: Influxdb connection: !ref influxdbConnection write: measurement: 'turbine' rotaryEncoderWrite: type: Cybus::Endpoint properties: protocol: Influxdb connection: !ref influxdbConnection write: measurement: 'rotary_encoder' rotary_encoder_angle: type: Cybus::Endpoint properties: protocol: Influxdb connection: !ref influxdbConnection subscribe: interval: 6000 query: > from(bucket:"turbine") |> range(start: -1d) |> filter(fn: (r) => r._measurement == "rotary_encoder") #---------------------------------------------------------------------------- # Mappings #---------------------------------------------------------------------------- # A mapping that overrides the measurement value with the name of the MQTT topic mappings: type: Cybus::Mapping properties: mappings: - subscribe: topic: 'turbine/#' publish: endpoint: !ref turbineWrite rules: - transform: # Add the topic as measurement expression: '$merge([$,{"measurement": $context.topic}])' # A mapping that will pass data from the MQTT topic to the write endpoint # allowing overriding the measurement by providing it in the topic - subscribe: topic: 'encoder/#' publish: endpoint: !ref rotaryEncoderWrite ``` {% endcode %} ## InfluxDB 3 Support Connectware integrates with InfluxDB 3 (Core, Enterprise, and Cloud) using the `Influxdb3` protocol. * [Connection Properties](/connectors/enterprise-connectors/influxdb/influxdb3connection.md) * [Endpoint Properties](/connectors/enterprise-connectors/influxdb/influxdb3endpoint.md) Set the `database` connection property to the InfluxDB 3 database that Connectware reads from and writes to. In InfluxDB 3, a database is the equivalent of a bucket in InfluxDB 2. ### Authentication InfluxDB 3 uses token-based authentication. Set the `token` connection property. Use a token that is scoped to the specific database and permission level you need, for example a write-only token, instead of an admin token. The `authScheme` connection property selects the token format: * Leave `authScheme` unset for **InfluxDB Cloud**. * Set `authScheme` to `Bearer` for **InfluxDB 3 Core** and **Enterprise**. To encrypt the connection with TLS, set `scheme` to `https` and make sure that the Connectware agent or pod trusts the certificate of the InfluxDB 3 server. For advanced TLS options, use the `transportOptions` connection property. ### Writing Data Define a `Cybus::Endpoint` with a `write` property to send data to InfluxDB 3. Publishing an MQTT message to the endpoint's `/set` topic writes one or more points. A message contains the fields to write and can also contain tags, a timestamp, and a measurement name that overrides the endpoint's default. Fields and tags that you define on the endpoint are merged with the values from the message. If a message does not contain a timestamp, the connector uses the current system time. Timestamps use millisecond precision (`precision: ms`). {% code lineNumbers="true" %} ```json { "measurement": "exhaust_temperature", "timestamp": 1752650000000, "tags": { "machine": "press_1", "line": "A" }, "fields": { "temperature": 81.5, "running": true, "status": "ok", "cycle_count": 1500 } } ``` {% endcode %} **Writing multiple points**: Send an array of data points in a single message to write multiple values at once. Writes are asynchronous. The connector buffers incoming points and flushes them as a single batch when either of two thresholds is met, whichever comes first: * The buffer reaches the `batchSize` limit (default: 1000 points). * The `flushInterval` elapses (default: 3000 milliseconds). If a write fails, the connector retries the data when the next batch is written. **Field data types.** InfluxDB 3 fixes each field column's data type on the first write to a measurement. By default, the connector infers numeric types from the payload — whole numbers become `integer`, other numbers become `float`. If a field can arrive as either, the type inferred from the first message may be rejected by later writes. To avoid this, declare types explicitly using the `fieldTypes` endpoint property. Allowed values are `float`, `integer`, `uinteger`, `string`, and `boolean`. ### Output Format on Write After writing data, you receive a confirmation message on the endpoint's `/res` topic: {% code lineNumbers="true" %} ```json { "value": true, "timestamp": 1784204563072 } ``` {% endcode %} * **timestamp**: Unix timestamp (in milliseconds) of the write. * **value**: Set to `true` when the write succeeds. ### Reading Data Define an endpoint with a `read` property for one-time queries, or a `subscribe` property for continuous polling. For subscriptions, set an `interval` in milliseconds or a `cronExpression` to control the polling schedule. Provide the query as the endpoint's `query` value. InfluxDB 3 uses **SQL** as its default query language. **InfluxQL** is also supported for compatibility with existing InfluxQL queries. Set `queryType` to `sql` (default) or `influxql`, either at the connection level or per endpoint. To bind values into a query, prefer the `params` endpoint property over string interpolation. Parameters supplied in a read request payload override the endpoint's defaults. ### Output Format on Read Connectware publishes query results as JSON to MQTT topics: * **One-time reads**: Results appear on the endpoint's `/res` topic. * **Subscriptions**: Results appear on the endpoint's default topic. You receive the data as a JSON array containing one object per result row. Each object contains the queried columns and the `time` value as a Unix timestamp in milliseconds. {% code lineNumbers="true" %} ```json [ { "cycle_count": 1500, "line": "A", "machine": "press_1", "running": true, "status": "ok", "temperature": 81.5, "time": 1752650000000, "value": null } ] ``` {% endcode %} ### InfluxDB 3 Service Commissioning File Example This example connects to an InfluxDB 3 server and writes data to the `machine_metrics` measurement, adding a constant tag to every point. {% file src="/files/McxY2URdo0qCXInMtZW1" %} {% code title="influxdb3-example.yml" lineNumbers="true" %} ```yaml description: | Sample InfluxDB 3 service commissioning file metadata: name: Cybus InfluxDB 3 Example provider: cybus homepage: https://cybus.io version: 1.0.0 parameters: influxHost: type: string description: 'Hostname or IP address of the InfluxDB 3 server' default: 'influxdb3' influxToken: type: string description: 'Authentication token for the InfluxDB 3 server' resources: influxdb3Connection: type: Cybus::Connection properties: protocol: Influxdb3 connection: host: !ref influxHost port: 8181 scheme: http authScheme: Bearer token: !ref influxToken database: machine_data machineMetricsWrite: type: Cybus::Endpoint properties: protocol: Influxdb3 connection: !ref influxdb3Connection write: measurement: machine_metrics tags: line: assembly-1 ``` {% endcode %} --- # Agent Instructions This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com. ## Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter: ``` GET https://docs.cybus.io/connectors/enterprise-connectors/influxdb.md?ask=&goal= ``` `ask` is the immediate question: it should be specific, self-contained, and written in natural language. `goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.