taosgen 参考手册

taosgen 是时序数据领域产品的性能基准测试工具，支持数据生成、写入性能测试等功能。taosgen 以“作业”为基础单元，作业是由用户定义，用于完成特定任务的一组操作集合。每个作业包含一个或多个步骤，并可通过依赖关系与其他作业连接，形成有向无环图（DAG）式的执行流程，实现灵活高效的任务编排。

taosgen 目前支持 Windows、Linux 和 macOS 系统。

taosgen 与 taosBenchmark 功能对比

taosgen 相比 taosBenchmark，具有以下优势和改进：

• 提供作业编排能力，作业支持 DAG 依赖关系，能模拟真实业务流程。
• 支持多种目标/协议（TDengine、MQTT、Kafka、InfluxDB），可用于数据库写入、消息发布等多种场景。
• 更丰富的数据生成方式。支持 lua 表达式生成数据，便于模拟真实业务数据。
• 支持即时数据生成，无需预先生成大批量数据文件，节省准备时间和模拟真实场景。
• 支持使用多种时间间隔策略控制数据写入操作，如根据数据产生的真实时间“播放”数据。
• 未知或错误配置项自动检测，能够及时发现并提示配置文件中的拼写错误或无效参数，提升配置安全性和易用性。
• 支持 TDengine 数据库连接池，能够高效管理和复用数据库连接资源。

taosgen 解决了 taosBenchmark 难以灵活配置、数据生成方式单一、扩展性不足等问题，更适合现代物联网、工业互联网的大数据测试需求。

工具获取

根据需要选择下载 taosgen 工具。 下载二进制发布包到本地，解压缩，为了便捷访问，可以创建符号链接存放到系统执行目录中，如 Linux 系统下执行命令：

tar zxvf taosgen-v0.8.6-linux-x64.tar.gz
cd taosgen
ln -sf `pwd`/taosgen /usr/bin/taosgen

运行

taosgen 支持通过命令行、配置文件指定参数配置，相同的参数配置，命令行优先级要高于配置文件。

提示

在运行 taosgen 之前，要确保所有待写入的目标 TDengine 集群已经在正常运行。

启动示例：

taosgen -h 127.0.0.1 -c config.yaml

命令行参数

命令行参数	功能说明
-h/--host	指定要连接的服务器的主机名称或 IP 地址，默认值为 localhost
-P/--port	指定要连接的服务器的端口号，默认值为 6030
-u/--user	指定用于连接服务器的用户名，默认为 root
-p/--password	指定用于连接服务器的密码，默认值为 taosdata
-c/--config-file	指定 yaml 格式配置文件的路径
-d/--log-dir	指定日志输出目录，默认值为 ./log
-o/--log-file	指定完整的日志文件路径（优先级高于 --log-dir），-f 已弃用
-v/--verbose	提高输出详细程度
-V/--version	显示版本信息并退出，不能与其它参数混用
-?/--help	显示帮助信息并退出

提示：当没有指定参数运行 taosgen 时，默认会创建 TDengine 数据库 tsbench、超级表 meters、1 万张子表，并为每张子表批量写入 1 万条数据。

配置文件参数

整体结构

配置文件分为："tdengine"、"mqtt"、"kafka"、"influxdb"、"schema"、"concurrency"、"jobs"、"ingress" 几部分。

• tdengine：描述 TDengine 数据库的相关配置参数。
• mqtt：描述 MQTT Broker 的相关配置参数。
• kafka：描述 Kafka Broker 的相关配置参数。
• influxdb：描述 InfluxDB 数据库的相关配置参数。
• schema：描述数据定义和生成的相关配置参数。
• concurrency：描述作业执行的并发度。
• jobs：列表结构，描述所有作业的具体相关参数。
• ingress：描述通用高效写入 gRPC 服务的相关配置参数。
• log_dir：指定日志输出目录，默认值为 log/。
• log_file：指定完整的日志文件路径（优先级高于 log_dir）。

作业的格式

作业（Job）是由用户定义并包含一组有序的步骤（steps）。每个作业具有唯一的作业标识符（即键名），并可指定依赖关系（needs），以控制与其他作业之间的执行顺序。作业的组成包括以下属性：

• 作业标识符（Job Key）：字符串类型，表示该作业在 jobs 列表中的唯一键名，用于内部引用和依赖管理。
• name：字符串类型，表示作业的显示名称，用于日志输出或 UI 展示。
• needs：列表类型，表示当前作业所依赖的其他作业的标识符列表。若不依赖任何作业，则为空列表。
• steps：列表类型，由一个或多个步骤（Step）组成，按顺序依次执行，定义了该作业的具体操作流程。 作业默认继承全局配置（如 tdengine、schema 等）。

步骤的格式

步骤（Step）是作业中基础的操作单位，代表某一种具体操作类型的执行过程。每个步骤按顺序运行，并可以引用预定义的 Action 来完成特定功能。步骤的组成包括以下属性：

• name：字符串类型，表示该步骤的显示名称，用于日志输出和界面展示。
• uses：字符串类型，指向要使用的 Action 路径或标识符，指示系统调用哪一个操作模块来执行此步骤。
• with：映射（字典）类型，包含传递给该 Action 的参数集合。参数内容因 Action 类型而异，支持灵活配置。 通过组合多个步骤，作业能够实现复杂的逻辑流程，例如 TDengine 创建超级表 & 子表、TDengine 写入数据等。

全局配置参数

日志参数

• log_dir（字符串）：指定日志输出目录，默认值为 log/。仅设置 log_dir 时，日志文件将写入 <log_dir>/taosgen.log。
• log_file（字符串）：指定完整的日志文件路径，设置后将覆盖 log_dir。

这些参数也可以通过命令行选项（--log-dir、--log-file）设置。优先级从高到低为：

1. --log-file（命令行）
2. --log-dir（命令行）
3. log_file（YAML 配置文件）
4. log_dir（YAML 配置文件）
5. 默认值：log/taosgen.log

TDengine 参数

• tdengine：描述 TDengine 数据库的相关配置参数，它包括以下属性：
  • dsn（字符串）：表示要连接的 TDengine 数据库的 DSN 地址，默认值为：taos+ws://root:taosdata@localhost:6041/tsbench。
  • drop_if_exists（布尔）：表示数据库已存在时是否删除该数据库，默认为 false。
  • props（字符串）：表示数据库支持的创建数据库的属性信息。 例如，precision ms vgroups 20 replica 3 keep 3650 分别设置了虚拟组数量、副本数及数据保留期限。
    • precision：指定数据库的时间精度，可选值为："ms"、"us"、"ns"。
    • vgroups：指定数据库的虚拟组的个数。
    • replica：指定数据库的副本格式。
  • pool：连接池配置，包含如下属性：
    • enabled（布尔）：表示是否启用连接池功能，默认值为 true，；
    • max_size（整型）：表示连接池的最大容量，默认值为 100；
    • min_size（整型）：表示连接池的最小容量，默认值为 2；
    • timeout（整型）：表示获取连接超时时间，单位毫秒，默认值为 1000；

MQTT 参数

• mqtt：描述 MQTT Broker 的相关配置参数，它包括以下属性：
  • uri（字符串）：MQTT Broker 的 uri 地址，默认值为 tcp://localhost:1883。
  • user（字符串）：登录 Broker 的用户名。
  • password（字符串）：登录 Broker 的密码。
  • client_id（字符串）：客户端唯一标识符前缀，默认值为 taosgen。
  • keep_alive（整数）：超时没有消息发送后会发送心跳，单位为秒，默认值为 5。
  • clean_session（布尔）：是否清除就会话状态，默认值为 true。
  • max_buffered_messages（整数）：客户端的最大缓冲消息数，默认值为 10000。

Kafka 参数

• kafka：描述 Kafka Broker 的相关配置参数，它包括以下属性：

  • bootstrap_servers (字符串)：Kafka 集群地址列表，格式为 "host:port"，多个地址用逗号分隔。

  • client_id（字符串）：客户端唯一标识符前缀，默认值为 taosgen。

  • topic (字符串)：指定要写入的 Kafka Topic 名称。

  • rdkafka_options（映射）：可指定底层 librdkafka 库支持的可选参数，如：security.protocol、sasl.mechanisms、sasl.username、sasl.password。

    • security.protocol (字符串)：指定客户端与 Kafka 集群之间通信的安全协议。可选值：
      • "plaintext"：明文传输，无加密（默认，若未配置）。
      • "ssl"：使用 SSL/TLS 加密通信。
      • "sasl_plaintext"：使用 SASL 进行身份验证，但通信为明文。
      • "sasl_ssl"：使用 SASL 进行身份验证，并使用 SSL/TLS 加密通信。
      • 默认值：未设置（即等效于 "plaintext"）。
    • sasl.mechanism (字符串)：当 security.protocol 设置为 "sasl_plaintext" 或 "sasl_ssl" 时，指定使用的 SASL 身份验证机制。常见可选值：
      • "PLAIN"：简单的用户名/密码验证，常用于外部身份提供商或基本认证。
      • "SCRAM-SHA-256"：基于挑战 - 响应的更安全机制，比 PLAIN 更安全。
      • "SCRAM-SHA-512"：比 SHA-256 更强的哈希算法。
      • "GSSAPI"：用于 Kerberos 认证。 注意：此字段必须与 security.protocol 同时配置，且其值取决于 Kafka Broker 端启用的 SASL 机制。
    • sasl.username (字符串)：SASL 身份验证的用户名。当使用 "PLAIN" 或 "SCRAM" 机制时需要提供。
    • sasl.password (字符串)：SASL 身份验证的密码。

    更多参数请参考 librdkafka 配置文档。

InfluxDB 参数

• influxdb：描述 InfluxDB 数据库的相关配置参数，它包括以下属性：
  • url（字符串）：InfluxDB 服务器的 HTTP 地址，默认值为 http://localhost:8086。
  • token（字符串）：用于身份验证的 API Token。也可通过环境变量 INFLUXDB_TOKEN 设置。
  • org（字符串）：InfluxDB 组织名称，默认值为 default。
  • bucket（字符串）：InfluxDB Bucket 名称，默认值为 default。

schema 参数

• schema：描述数据定义和生成模式的相关配置参数。

  • name（字符串）：schema 的名称。

  • from_csv：描述 CSV 文件作为数据源时的相关配置参数。

    • tags：描述 tags 的配置参数。
      • file_path（字符串）：标签数据 CSV 文件路径。支持单个文件、目录路径（自动识别目录下所有 .csv 文件）和 glob 通配符（如 tags_*.csv）。多文件按文件名字母升序读取，所有文件须具有相同的列结构。
      • has_header（布尔）：是否包含表头行，默认为 true。
      • tbname_index（整数）：指定表名称所在的列索引（从 0 开始），默认为 -1，表示未生效。
      • exclude_indices（字符串）：如果仅想使用部分标签列时，此参数用于指定剔除的无用标签列的索引（从 0 开始），列索引之间使用英文逗号，分隔，默认值为空，表示不剔除。
    • columns：时序数据列的配置参数。
      • loading_mode（字符串）：CSV 数据加载模式，可选值为 "preload"（全量加载到内存）和 "streaming"（流式逐行读取），默认为 "preload"。
      • file_path（字符串）：时序数据 CSV 文件路径。支持单个文件、目录路径（自动识别目录下所有 .csv 文件）和 glob 通配符（如 data_*.csv）。多文件按文件名字母升序读取并逻辑串联为连续数据流，所有文件须具有相同的列结构。
      • has_header（布尔）：是否包含表头行，默认为 true。
      • repeat_read（布尔）：是否重复读取数据，默认为 false。
      • tbname_index（整数）：指定子表名称所在的列索引（从 0 开始），默认为 -1，表示未生效。
      • timestamp_index（整数）：指定时间戳列的索引（从 0 开始），默认为 -1，表示未生效。
      • timestamp_precision（字符串）：表示时间戳列的时间精度，可选值为 "s"、"ms"、"us"、"ns"。
      • timestamp_offset：描述时间戳数值的偏移配置参数。
        • offset_type（字符串）：表示时间戳偏移类型，可选值为："relative"、"absolute"。
        • value（字符串或整型）：表示时间戳的偏移量（relative）或起始时间戳（absolute）：
          • 时间戳偏移类型为 "relative" 时：字符串类型，格式为 ±[数值][单位] 组合（示例："+1d3h30m" 表示加 1 天 3 小时 30 分钟），支持以下时间单位：
            • y：年偏移量
            • M：月偏移量（大写）
            • d：天偏移量
            • h：小时偏移量
            • m：分钟偏移量（小写）
            • s：秒偏移量
          • 时间戳偏移类型为 "absolute" 时：整型或字符串类型，格式如下：
            • 时间戳数值（精度由 timestamp_precision 参数决定）
            • ISO 8601 格式字符串（"YYYY-MM-DD HH:mm:ss"）

  • tbname：描述生成表名称的相关配置参数：

    • prefix（字符串）：表名前缀，默认为 "d"。
    • count（整数）：要创建的表数量，默认为 10000。
    • from（整数）：表名称的起始下标（包含），默认为 0。

  • tags（列表）：描述表标签列结构的模式定义。默认配置为：groupid INT, location VARCHAR(24)。

  • columns（列表）：描述表普通列结构的模式定义。默认配置为：ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT。

  • generation：描述数据生成行为相关的配置参数。

    • interlace（整数）：控制交错方式生成表数据的行数，默认值为 0，表示不启用交错模式。
    • concurrency（整数）：表示生成数据的线程数量，默认值为写入线程数量。
    • rows_per_table（整数），每个数据表写入的行数，默认值为 10000，-1 表示无限数据。
    • rows_per_batch（整数），表示每次批量请求写入的最大行数，默认值为 10000。
    • num_cached_batches（整数），表示提前生成数据并缓存批量请求的数量，0 表示关闭数据缓存，默认值为 10000。
    • tables_reuse_data（布尔）：多表是否复用相同数据，默认为 true。

列配置包含属性

每列包含以下属性：

• name（字符串）：表示列的名称，当 count 属性大于 1 时，name 表示的是列名称前缀，比如：name：current，count：3，则 3 个列的名字分别为 current1、current2、current3。

• type（字符串）：表示数据类型，支持以下类型（不区分大小写，与 TDengine 的数据类型兼容）：

  • 整型：timestamp、bool、tinyint、tinyint unsigned、smallint、smallint unsigned、int、int unsigned、bigint、bigint unsigned。
  • 浮点型：float、double、decimal。
  • 字符型：nchar、varchar（binary）。

  目前，还不支持以下数据类型：json、geometry、varbinary、decimal、blob。

• count（整数）：表示指定该类型的列连续出现的数量，例如 count：4096 即可生成 4096 个指定类型的列。

• props（字符串）：表示 TDengine 数据库的列支持的属性信息，可以包含以下属性：

  • encode：指定此列两级压缩中的第一级编码算法。
  • compress：指定此列两级压缩中的第二级加密算法。
  • level：指定此列两级压缩中的第二级加密算法的压缩率高低。

• gen_type（字符串）：指定此列生成数据的方式，默认值为 random，支持的类型有：

  • random：随机方式生成。
  • order：按自然数顺序增长，仅适用整数类型。
  • expression：根据表达式生成。适用整数类型、浮点数类型 float、double 和字符类型。

• null_ratio（浮点数）：指定生成 NULL 值的比例，取值范围 [0.0, 1.0]，默认值为 0.0。NULL 表示空值，写入后替换该列的最新值为空。

• none_ratio（浮点数）：指定生成 NONE 值的比例，取值范围 [0.0, 1.0]，默认值为 0.0。NONE 表示数据缺失，写入后不替换该列的最新值，保留原有值。null_ratio 与 none_ratio 之和不得超过 1.0。

数据生成方式详解

• random：随机方式生成

  • distribution（字符串）：表示随机数的分别模型，目前仅支持均匀分布，后续按需扩充，默认值为 "uniform"。
  • min（浮点数）：表示列的最小值，仅适用整数类型和浮点数类型，生成的值将大于或等于最小值。
  • max（浮点数）：表示列的最大值，仅适用整数类型和浮点数类型，生成的值将小于最大值。
  • values（列表）：指定随机数据的取值范围，生成的数据将从中随机选取。
  • min_length（整数）：仅适用于变长字符类型（nchar、varchar 或 binary），指定生成字符串的最小长度。这里的“容量”指类型定义的最大长度。生效规则如下：若 min_length 和 max_length 都未设置，则生成长度固定为类型定义的最大长度；若仅设置 min_length，则 max_length 默认为类型定义的最大长度；若同时设置，则需满足 0 ≤ min_length ≤ max_length ≤ 容量。设置后每次生成的字符串长度会在 min_length 到 max_length 之间随机变化。
  • max_length（整数）：仅适用于变长字符类型（nchar、varchar 或 binary），指定生成字符串的最大长度。这里的“容量”指类型定义的最大长度。若仅设置 max_length，则 min_length 默认为 0；若同时设置 min_length 和 max_length，则需满足 0 ≤ min_length ≤ max_length ≤ 容量。

• order：按自然数顺序增长，仅适用整数类型，达到最大值后会自动翻转到最小值

  • min（整数）：表示列的最小值，生成的值将大于或等于最小值。
  • max（整数）：表示列的最大值，生成的值将小于最大值。

• expression：根据表达式生成。适用于整数类型、浮点类型和字符类型。如果未显式指定 gen_type，但检测到包含 expr 属性，则会自动设置 gen_type 为 expression。

  • expr（字符串）：表示生成数据的表达式内容，表达式语法采用 lua 语言，内置变量：
    • _i 表示调用索引，从 0 开始，如："2 + math.sin（_i/10）"；
    • _table 表示该表达式为哪张表构建数据；
    • _last 表示该表达式上次返回的数值，仅对数值类型生效，初始值为 0.0；

  为了说明表达式方式的数据描述能力，下面举一个更复杂的表达式样例：

  （math.sin（_i / 7） * math.cos (_i / 13) + 0.5 *（math.random（80, 120） / 100）） *（（_i % 50 < 25） and（1 + 0.3 * math.sin (_i / 3)) or 0.7) + 10 *(math.floor (_i / 100) % 2)

  它结合了多种数学函数、条件逻辑、周期性行为和随机扰动，模拟一个非线性、带噪声、分段变化的动态数据生成过程，组成部分（A + B）× C + D。功能分解说明：

  部分	内容	类别	作用
  A	math.sin（_i / 7） * math.cos（_i / 13）	基础信号	双频调制，生成复杂波形（拍频效应）
  B	0.5 *（math.random（80, 120） / 100）	噪声	添加 80%~120% 的随机扰动（模拟噪声）
  C	（（_i % 50 < 25） and（1 + 0.3 * math.sin（_i / 3）） or 0.7）	动态增益调制	每 50 次调用切换一次增益（前 25 次高增益，后 25 次低增益）
  D	10 *（math.floor（_i / 100） % 2）	基线阶跃变化	每 100 次调用切换一次基线（0 或 10），模拟阶跃变化，表示高峰/低谷

ingress 参数

ingress 是 taosgen 提供的通用高效数据写入 gRPC 服务。启用后，taosgen 按配置文件中固定的 TDengine 连接、超级表和 schema 启动服务，外部请求只传 数据值，不传数据库名、超级表名、列名列表、列类型、DDL 或 SQL。

启动方式与普通配置命令一致：

taosgen -c ingress.yaml

运行模式如下：

配置内容	行为
ingress.enabled=true，jobs 省略或为空	启动 gRPC 服务，长运行
ingress.enabled=true 且 jobs 非空	先执行 startup jobs，成功后启动 gRPC 服务
jobs 非空，ingress 缺省或为 false	执行现有 jobs，完成后退出
jobs 为空且 ingress 缺省或为 false	无可执行任务，启动失败并提示配置错误

一个 ingress 实例只绑定一个 TDengine 数据库、一个超级表和一个固定 schema。 如需写入多个数据库或多个超级表，应启动多个 taosgen ingress 实例并分别配置 不同监听端口。

注意

当前第一版 ingress 未提供 TLS 或鉴权，必须仅部署在可信内网或本机，禁止将 listen 暴露到不可信网络。如需对外提供服务，应在受控网络内或前置安全网关。

最小配置示例：

tdengine:
  dsn: taos://root:taosdata@127.0.0.1:6030/power
  props: precision 'ms'

schema:
  name: meters
  columns:
    - name: ts
      type: timestamp
      precision: ms
    - name: current
      type: double
  tags:
    - name: device_id
      type: binary(64)

ingress:
  enabled: true
  listen: 127.0.0.1:7600
  default_ack_mode: enqueue_only
  concurrency: 4
  auto_create_table: false

主要配置项：

• enabled（布尔）：是否启用 ingress 服务，默认值为 false。
• listen（字符串）：gRPC 监听地址，默认值为 127.0.0.1:7600。
• service_concurrency（整数）：gRPC 接收侧并发度，默认值为当前机器硬件并发数。
• max_message_bytes（整数）：单个 gRPC 请求最大消息大小，默认值为 4194304。
• rows_per_batch（整数）：单请求最大行数；未配置时从 schema.generation.rows_per_batch 推导，该字段默认值为 10000。
• max_tables_per_request（整数）：单请求最大子表数；未配置时根据 rows_per_batch 与每表行数推导，普通模式使用 schema.generation.rows_per_table，interlace 模式使用 schema.generation.interlace 行数。
• default_ack_mode（字符串）：请求未指定 ack_mode 时的默认回执模式，默认值为 enqueue_only。
• request_timeout_ms（整数）：等待队列空位的超时时间，默认值为 30000；设置为 0 表示不等待，立即尝试入队。
• queue_capacity（整数）：内部队列容量系数；有效队列容量为 queue_capacity * concurrency，默认值为 10。
• concurrency（整数）：TDengine STMT2 写入并发度，默认值为 8。
• dry_run（布尔）：是否只验证接收链路而不真实写入 TDengine，默认值为 false。
• auto_create_table（布尔）：是否使用 STMT2 自动建子表写入，默认值为 false。
• writer.failure_handling：写入失败处理策略，包含 max_retries、retry_interval_ms、on_failure。

对外提供以下 gRPC 接口：

• Ingest：写入一批 TableBatch 或 InterlaceBatch 数据，返回回执阶段、接收行数和队列深度。
• Health：查询服务是否可用。
• GetStats：查询接收行数、写入行数、失败行数、队列深度和在途请求数等统计。

数据组织形式：

• TableBatch：按子表组织，同一子表批次内 table_name 和 tags 只传一次，时间戳和普通列使用列式数组传输。
• InterlaceBatch：请求级传输表名集合、第一时间戳列和普通列数组，数组按 offset = table_index * rows_per_table + row_index 展开；连续表名可使用 TableNameRange(prefix,start,count,step)。

注意：

• 当前仅支持 accept_only 和 enqueue_only；accept_only 不入队、不写 TDengine。
• 是否真实写入 TDengine 由 dry_run 决定，不由 ack_mode 决定。
• auto_create_table=false 时请求不得携带 tags，子表需提前创建。
• InterlaceBatch 第一版仅支持 auto_create_table=false。

行动的种类

行动（Action）是封装好的可复用操作单元，用于完成特定功能。每个行动代表一类独立的操作逻辑，可以在不同的步骤（Step）中被调用和执行。通过将常用操作抽象为标准化的行动模块，系统实现了良好的扩展性与配置灵活性。 同一类型的行动可以在多个步骤中并行或重复使用，从而支持多样化的任务流程编排。例如：创建数据库、定义超级表、生成子表、写入数据等核心操作，均可通过对应的行动进行统一调度。 目前系统支持以下内置行动：

• tdengine/create-database：用于创建 TDengine 数据库
• tdengine/create-super-table：用于创建 TDengine 超级表
• tdengine/create-child-table：用于创建 TDengine 超级表的子表
• tdengine/insert：用于向指定的 TDengine 数据库中写入数据
• mqtt/publish：用于向指定的 MQTT Broker 发布数据
• kafka/produce：用于向指定的 Kafka Broker 发布数据
• influxdb/write：用于向指定的 InfluxDB 数据库写入数据 每个行动在调用时可通过 with 字段传入参数，具体参数内容因行动类型而异。

备注

tdengine/insert-data 为 v0.7.x 及更早版本的旧名称，v0.8.0 起继续使用会收到提示："Action 'tdengine/insert-data' is deprecated and will be removed in future versions. Please use 'tdengine/insert' instead"。该名称自 v0.8.3 起不再支持（对应 TDengine 3.3.6.39/3.3.8.16/3.4.0.2）。

创建 TDengine 数据库行动的格式

tdengine/create-database 行动用于在指定的 TDengine 数据库服务器上创建一个新的数据库。通过配置的连接信息和数据库参数，用户可以轻松地定义新数据库的各种属性，如数据库名称、是否在存在时删除旧数据库、时间精度等。

• checkpoint：描述写入数据中断/恢复功能相关配置参数：
  • enabled：是否开启写入数据中断/恢复功能。
  • interval_sec：数据写入进度进行存储间隔，单位为秒。

创建 TDengine 超级表行动的格式

tdengine/create-super-table 行动用于在指定数据库中创建一个新的超级表。通过传递必要的连接信息和超级表配置参数，用户能够定义超级表的各种属性，如表名、普通列和标签列等。

• schema：默认使用全局的 schema 配置信息，当需要差异化时可在此行动下单独定义。

创建 TDengine 子表行动的格式

tdengine/create-child-table 行动用于基于指定的超级表，在目标数据库中批量创建多张子表。每张子表可以拥有不同的名称和标签列数据，从而实现对时间序列数据的有效分类与管理。该行动支持从生成器（Generator）或 CSV 文件两种来源定义子表名称及标签列信息，具备高度灵活性和可配置性。

• schema：默认使用全局的 schema 配置信息，当需要差异化时可在此行动下单独定义。
• batch：控制批量创建子表时的行为：
  • size（整数）： 每批创建的子表数量，默认值为 1000。
  • concurrency（整数）：并发执行的批次数量，提升创建效率，默认值为 10。

写入 TDengine 数据行动的格式

tdengine/insert 行动用于将数据写入到指定的子表中。它支持从生成器或 CSV 文件两种来源获取子表名称、普通列数据，并允许用户通过多种时间戳策略控制数据的时间属性。此外，还提供了丰富的写入控制策略以优化数据写入过程，具备高度灵活性和可配置性。

• schema：默认使用全局的 schema 配置信息，当需要差异化时可在此行动下单独定义。
• format（字符串）：描述数据写入时使用的格式，可选值为：sql、stmt、schemaless，默认使用 stmt。
  • sql：使用 SQL 语句写入数据。
  • stmt：使用参数化写入（Prepared Statement）方式写入数据，适合高性能批量写入场景。
  • schemaless：使用行协议（Line Protocol）方式写入数据，无需预先创建超级表和子表，适合模拟 Telegraf 等采集器向 TDengine 发送数据的场景。
• auto_create_table（布尔）：表示是否使用 TDengine 自动建表功能在写入数据时动态创建表，默认值为 false。
• tbname_key（字符串）：仅在 schemaless 格式下生效，用于指定行协议（Line Protocol）输出中代表子表名称的 tag key。如果此参数被设置为空字符串（""），则不在行协议中输出子表名称 tag。默认值为 ""。
• concurrency（整数）：并发写入数据的线程数量，默认值为 8。
• failure_handling：表示失败处理策略：
  • max_retries（整数）：最大重试次数，默认值为 0。
  • retry_interval_ms（整数）：重试间隔，单位为毫秒，默认值为 1000，仅在 max_retries > 0 时有效。
  • on_failure（字符串）：默认值为 exit，表示失败后的行为，可选值为：
    • exit：失败后自动退出程序
    • skip：失败后警告用户并跳过继续执行
• time_interval：控制写入过程中时间间隔分布策略。
  • enabled（布尔）：表示是否启用时间间隔控制，默认值为 false。
  • interval_strategy（字符串）：表示时间间隔策略类型，默认值为 fixed。可选值为：
    • fixed：固定的时间间隔。
    • first_to_first：本次发送数据的首行的时间列 - 上次发送数据的首行的时间列。
    • last_to_first：本次发送数据的首行的时间列 - 上次发送数据的末行的时间列。
    • literal：根据本次发送数据的首行的时间列的值的时间点来发送，模拟实时产生数据的场景。
  • fixed_interval：仅在 interval_strategy = fixed 时生效：
    • base_interval（整数）：表示固定间隔数值，单位为毫秒，默认值为 1000。
  • dynamic_interval：仅在 interval_strategy = first_to_first / last_to_first 时生效：
    • min_interval（整数）：表示最小时间间隔阈值，单位为毫秒，默认值为 -1，表示未生效。
    • max_interval（整数）：表示最大时间间隔阈值，单位为毫秒，默认值为 -1，表示未生效。
  • wait_strategy（字符串）：表示在开启时间间隔控制时，发送写入请求之间的等待策略，默认值为：sleep，可选值为：
    • sleep：睡眠，归还当前线程的执行权给操作系统。
    • busy_wait：忙等待，保持当前线程的执行权。
• checkpoint：描述写入数据中断/恢复功能相关配置参数（目前仅支持 stmt 格式、生成器方式数据源）：
  • enabled：是否开启写入数据中断/恢复功能。
  • interval_sec：数据写入进度进行存储间隔，单位为秒。

发布 MQTT 数据行动的格式

mqtt/publish 行动用于将数据发布到指定的 topic 中。它支持从生成器或 CSV 文件两种来源获取数据，并允许用户通过多种时间戳策略控制数据的时间属性。此外，还提供了丰富的写入控制策略以优化数据发布过程，具备高度灵活性和可配置性。

• schema：默认使用全局的 schema 配置信息，当需要差异化时可在此行动下单独定义。
• format（字符串）：描述数据发布时使用的格式，目前仅支持 json，默认值为 json。
• concurrency（整数）：并发发布数据的线程数量，默认值为 8。
• failure_handling：参数说明请参考 写入 TDengine 数据行动的格式 中的同名参数。
• time_interval：参数说明请参考 写入 TDengine 数据行动的格式 中的同名参数。
• topic（字符串）：要发布消息的 MQTT Topic，默认值为 tsbench/{table}。支持通过占位符语法发布到动态主题，占位符语法如下：
  • {table}：表示表名数据
  • {column}：表示列数据，column 是列字段名称
• qos（整数）：QoS 等级，取值范围为 0、1、2，默认为 0。
• retain（布尔）：MQTT Broker 是否保留最后一条消息，默认值为 false。
• tbname_key (字符串)：用于指定输出中代表表名的字段名称。如果此参数被设置为空字符串（""），则不输出表名信息。默认值为 ""。
• records_per_message（整数）：每条消息包含的记录数，默认为 1。

发布 Kafka 数据行动的格式

kafka/produce 行动用于将数据发布到指定的 topic 中。它支持从生成器或 CSV 文件两种来源获取数据，并允许用户通过多种时间戳策略控制数据的时间属性。此外，还提供了丰富的写入控制策略以优化数据发布过程，具备高度灵活性和可配置性。

• schema：默认使用全局的 schema 配置信息，当需要差异化时可在此行动下单独定义。
• concurrency（整数）：并发发布数据的线程数量，默认值为 8。
• failure_handling：参数说明请参考 写入 TDengine 数据行动的格式 中的同名参数。
• time_interval：参数说明请参考 写入 TDengine 数据行动的格式 中的同名参数。
• key_pattern (字符串)：消息 Key 的组成模式，支持通过占位符语法生成动态 key，默认值为 {table}，占位符语法如下：
  • {table}：表示表名数据
  • {column}：表示列数据，column 是列字段名称
• key_serializer (字符串)：消息 Key 的序列化方式，支持 "string-utf8"、"int8"、"uint8"、"int16"、"uint16"、"int32"、"uint32"、"int64"、"uint64"，默认为 "string-utf8"，控制如何将 key_pattern 解析后的结果序列化为 Kafka 消息的 key 字节流。
  • "string-utf8": 将模板替换后的结果视为字符串，直接以 UTF-8 编码生成字节流。
  • 整数：将字段模板替换后的结果解析为整数。仅支持单个字段占位符，序列化时使用该整数类型，并以大端序（big-endian）格式编码为二进制数据发送。
• value_serializer (字符串)：消息 Value 的序列化方式，支持 "json"、"influx"，默认为 "json"。
• acks (字符串)：生产者确认机制设置，如 "all"、"1"、"0"，默认为 "0"；
  • "all"：生产者必须等待 ISR（In-Sync Replicas，同步副本集）中的所有副本都成功接收到消息并将其写入本地日志后，才会认为消息发送成功。
  • "1"：生产者只需要等待分区 Leader 副本成功接收到消息并将其写入本地日志（Log），就会认为消息发送成功，并立即向应用程序返回确认。
  • "0"：生产者完全不等待任何确认。一旦消息被成功发送到网络（甚至只是放入了生产者的发送缓冲区），就立即认为发送成功。
• compression (字符串)：消息压缩类型，支持 "none"、"gzip"、"snappy"、"lz4"、"zstd"，默认为 "none"。
• tbname_key (字符串)：用于指定输出中代表表名的字段名称。在 json 格式中作为 JSON 字段名，在 influx 格式中作为行协议的 tag key。如果此参数被设置为空字符串（""），则不输出表名信息。默认值为 ""。
• records_per_message（整数）：每条消息包含的记录数，默认为 1。

写入 InfluxDB 数据行动的格式

influxdb/write 行动用于将数据通过 InfluxDB v2 Write API 以行协议（Line Protocol）格式写入到指定的 InfluxDB Bucket 中。它支持从生成器或 CSV 文件两种来源获取数据，并允许用户通过多种时间戳策略控制数据的时间属性。此外，还提供了丰富的写入控制策略以优化数据写入过程，具备高度灵活性和可配置性。

• schema：默认使用全局的 schema 配置信息，当需要差异化时可在此行动下单独定义。schema 中的 name 字段将作为 InfluxDB 的 measurement 名称。
• concurrency（整数）：并发写入数据的线程数量，默认值为 8。
• failure_handling：参数说明请参考 写入 TDengine 数据行动的格式 中的同名参数。
• time_interval：参数说明请参考 写入 TDengine 数据行动的格式 中的同名参数。
• precision（字符串）：时间戳精度，可选值为 "ns"、"us"、"ms"、"s"，默认为 "ns"。
• batch_size（整数）：每次 HTTP 请求包含的行协议行数，默认为 5000。InfluxDB 官方推荐值为 5000。
• gzip（布尔）：是否对 HTTP 请求体启用 gzip 压缩，默认为 false。启用后可显著减少网络带宽占用。
• tbname_key（字符串）：用于指定行协议（Line Protocol）输出中代表表名称的 tag key。如果此参数被设置为空字符串（""），则不在行协议中输出表名称 tag。默认值为 ""。

配置文件示例

生成器方式生成数据 STMT 方式写入 TDengine 示例

该示例展示了如何使用 taosgen 工具模拟一万台智能电表，每台智能电表采集电流、电压、相位三个物理量，它们每隔 5 分钟产生一条记录，电流的数据用随机数，电压用正弦波模拟，产生的这些数据采用 WebSocket 的方式写入 TDengine 的 tsbench 数据库的超级表 meters。

配置详解：

• TDengine 配置参数
  • 连接信息：通过 DSN 定义数据库连接相关参数。
  • 数据库的属性：定义是否重新创建数据库，设置时间精度为毫秒，4 个 vgroup。
• schema 配置参数
  • 名称：指定超极表的名称。
  • 子表名称：定义生成一万张子表名称的规则，格式为 d0 到 d9999。
  • 超级表字段结构信息：定义超级表结构，包含 3 个普通列（电流、电压、相位）和 2 个标签列（组 ID、位置）。
    • 时间戳：配置了时间戳生成策略，从指定时间戳 1700000000000 (2023-11-14 22:13:20 UTC) 开始，以 5 分钟的步长递增。
    • 时序数据：current 和 phase 使用指定范围的随机数，voltage 使用正弦波模拟。
    • 标签数据：groupid 和 location 使用指定范围的随机数。
  • 数据生成行为：使用交错模式写入，每张子表写入 1 万条记录，每批写入请求最大行数为 1 万行。
• 创建子表：指定 10 线程并发创建子表，每批发送 1000 张子表创建请求。
• 数据写入：使用默认的 stmt (参数化写入) 格式、8 线程并发写入数据，极大提升了批量数据写入的性能。

场景说明：

此配置专为 TDengine 数据库的性能基准测试 而设计。它适用于模拟大规模物联网设备（如电表、传感器）持续产生高频数据的场景，用于：

• 测试和评估 TDengine 集群在海量时间序列数据写入压力下的吞吐量、延迟和稳定性。
• 验证数据库 schema 设计、资源规划以及不同硬件配置下的性能表现。
• 为工业物联网等领域的系统容量规划提供数据支撑。

tdengine:
  dsn: taos+ws://root:taosdata@127.0.0.1:6041/tsbench
  drop_if_exists: true
  props: precision 'ms' vgroups 4

schema:
  name: meters
  tbname:
    prefix: d
    count: 10000
    from: 0
  columns:
    - name: ts
      type: timestamp
      start: 1700000000000
      precision : ms
      step: 300s
    - name: current
      type: float
      min: 0
      max: 100
    - name: voltage
      type: int
      expr: '220 * math.sqrt(2) * math.sin(_i)'
    - name: phase
      type: float
      min: 0
      max: 360
  tags:
    - name: groupid
      type: int
      min: 1
      max: 10
    - name: location
      type: binary(24)
      values:
        - New York
        - Los Angeles
        - Chicago
        - Houston
        - Phoenix
        - Philadelphia
        - San Antonio
        - San Diego
        - Dallas
        - Austin
  generation:
    interlace: 1
    rows_per_table: 10000
    rows_per_batch: 10000
    num_cached_batches: 0

jobs:
  # TDengine insert job
  insert:
    steps:
      - uses: tdengine/create-super-table
      - uses: tdengine/create-child-table
        with:
          batch:
            size: 1000
            concurrency: 10

      - uses: tdengine/insert
        with:
          concurrency: 8

查看源码

其中，tdengine、schema::name、schema::tbname、schema::tags、tdengine/create-child-table::batch、tdengine/insert::concurrency 可以使用默认值，进一步简化配置。

schema:
  columns:
    - name: ts
      type: timestamp
      start: 1700000000000
      precision : ms
      step: 300s
    - name: current
      type: float
      min: 0
      max: 100
    - name: voltage
      type: int
      expr: '220 * math.sqrt(2) * math.sin(_i)'
    - name: phase
      type: float
      min: 0
      max: 360
  generation:
    interlace: 1
    num_cached_batches: 0

jobs:
  # TDengine insert job
  insert:
    steps:
      - uses: tdengine/create-super-table
      - uses: tdengine/create-child-table
      - uses: tdengine/insert

查看源码

CSV 文件方式生成数据 STMT 方式写入 TDengine 实例

该示例展示了如何使用 taosgen 工具模拟一万台智能电表，每台智能电表采集电流、电压、相位三个物理量，它们每隔 5 分钟产生一条记录，测点数据读取自 CSV 文件，采用 WebSocket 的方式写入 TDengine 的 tsbench 数据库的超级表 meters。

配置详解：

• TDengine 配置参数
  • 连接信息：通过 DSN 定义数据库连接相关参数。
  • 数据库的属性：定义是否重新创建数据库，设置时间精度为毫秒，4 个 vgroup。
• schema 配置参数
  • 名称：指定超极表的名称。
  • from_csv 配置定义了子表名称、标签列和时序数据列的来源。
    • 子表名称：使用文件 ctb-tags.csv 中索引为 2 的列。
    • 标签数据：使用文件 ctb-tags.csv 中排除子表名称列之外的所有列。
    • 时间戳：使用文件 ctb-data.csv 中索引为 1 的列，且在原始数据基础上增加 10 秒。
    • 时序数据：使用文件 ctb-data.csv 中的数据，该文件第 0 列为子表名，用于关联数据与子表。。
  • 超级表字段结构信息：定义超级表结构，包含 3 个普通列（电流、电压、相位）和 2 个标签列（组 ID、位置）。
  • 数据生成行为：使用交错模式写入，每张子表写入 1 万条记录，每批写入请求最大行数为 1 万行。
• 创建子表：指定 10 线程并发创建子表，每批发送 1000 张子表创建请求。
• 数据写入：使用默认的 stmt (参数化写入) 格式、8 线程并发写入数据，极大提升了批量数据写入的性能。

场景说明：

此配置专为从现有 CSV 文件导入设备元数据和历史数据到 TDengine 数据库而设计。它适用于以下场景：

• 数据迁移：将已收集存储于 CSV 文件中的设备元数据（标签）和历史监测数据迁移至 TDengine 数据库。
• 系统初始化：为新的监控系统初始化一批设备及其历史数据，用于系统测试、演示或回溯分析。
• 数据回放：通过重新注入历史数据，模拟实时数据流，用于测试系统处理能力或重现特定历史场景。

tdengine:
  dsn: taos+ws://root:taosdata@127.0.0.1:6041/tsbench
  drop_if_exists: true
  props: precision 'ms' vgroups 4

schema:
  name: meters
  from_csv:
    tags:
      file_path: ./conf/ctb-tags.csv
      tbname_index: 2
      exclude_indices:
    columns:
      file_path: ./conf/ctb-data.csv
      tbname_index : 0
      timestamp_index: 1
      timestamp_precision: ms
      timestamp_offset:
        offset_type: relative
        value: +10s
      repeat_read: false
  columns:
    - name: ts
      type: timestamp
      precision : ms
    - name: current
      type: float
    - name: voltage
      type: int
    - name: phase
      type: float
  tags:
    - name: groupid
      type: int
    - name: location
      type: binary(24)
  generation:
    interlace: 1
    rows_per_table: 10000
    rows_per_batch: 10000
    num_cached_batches: 0
    tables_reuse_data: false

jobs:
  # TDengine insert job
  insert:
    steps:
      - uses: tdengine/create-super-table
      - uses: tdengine/create-child-table
        with:
          batch:
            size: 1000
            concurrency: 10
      - uses: tdengine/insert
        with:
          concurrency: 8

查看源码

其中：

• ctb-tags.csv 文件内容格式为：

groupid,location,tbname
1,California.Campbell,d1
2,Texas.Austin,d2
3,NewYork.NewYorkCity,d3

• ctb-data.csv 文件内容格式为：

tbname,ts,current,voltage,phase
d1,1700000010000,5.23,221.5,146.2
d3,1700000010000,8.76,219.8,148.7
d2,1700000010000,12.45,223.1,147.3
d3,1700000310000,9.12,220.3,149.1
d2,1700000310000,11.87,222.7,145.8
d1,1700000310000,4.98,220.9,147.9

生成器方式生成数据并发布数据到 MQTT Broker 示例

该示例展示了如何使用 taosgen 工具模拟一万台智能电表，每台智能电表采集电流、电压、相位、位置四个物理量，它们每隔 5 分钟产生一条记录，电流的数据用随机数，电压用正弦波模拟，产生的这些数据通过 MQTT 协议进行发布。

配置详解：

• MQTT 配置参数

  • 连接信息：使用 URI 描述连接 MQTT Broker 的信息。

• schema 配置参数

  • 名称：指定 schema 的名称。
  • 表名称：定义生成一万张逻辑表的名称规则，格式为 d0 到 d9999，用于组织和标识生成的数据。
  • 表字段结构信息：定义数据表结构，包含 4 个普通列（电流、电压、相位、设备位置）。
    • 时间戳：配置了时间戳生成策略，从指定时间戳 1700000000000 (2023-11-14 22:13:20 UTC) 开始，以 5 分钟的步长递增。
    • 时序数据：current、phase 和 location 使用指定范围的随机数，voltage 使用正弦波模拟。
  • 数据生成行为：使用交错模式写入，每张表写入 1 万条记录，每批写入请求最大行数为 1 万行。

• 数据发布：使用 8 线程并发向 MQTT Broker 主题发布数据，提高吞吐量。

  • 主题配置 (topic): 使用动态主题 factory/{table}/{location}，其中：
    • {table} 占位符将被实际生成的子表名称替换。
    • {location} 占位符将被生成的 location 列值替换，实现按设备位置发布到不同主题。
  • qos：服务质量等级设置为 1（至少交付一次）。

场景说明：

此配置专为向 MQTT 消息代理发布模拟设备数据而设计。它适用于以下场景：

• MQTT 消费者测试：模拟大量设备向 MQTT 代理发布数据，用于测试 MQTT 消费者端的处理能力、负载均衡和稳定性。
• 物联网平台演示：快速构建一个模拟的物联网环境，展示设备数据如何通过 MQTT 协议接入平台。
• 规则引擎测试：结合 MQTT 主题的动态特性（如按设备位置路由），测试基于 MQTT 的主题订阅和消息路由规则。
• 实时数据流模拟：模拟实时产生的设备数据流，用于测试流处理框架的数据消费和处理能力。

mqtt:
  uri: tcp://localhost:1883
  user: root
  password: taosdata

schema:
  name: meters
  tbname:
    prefix: d
    count: 10000
    from: 0
  columns:
    - name: ts
      type: timestamp
      start: 1700000000000
      precision : ms
      step: 300s
    - name: current
      type: float
      min: 0
      max: 100
    - name: voltage
      type: int
      expr: '220 * math.sqrt(2) * math.sin(_i)'
    - name: phase
      type: float
      min: 0
      max: 360
    - name: location
      type: varchar(20)
      values:
        - Chicago
        - Houston
        - Phoenix
        - Philadelphia
        - Dallas
        - Austin
  generation:
    interlace: 1
    concurrency: 8
    rows_per_table: 10000
    rows_per_batch: 10000
    num_cached_batches: 0

jobs:
  # MQTT publish job
  publish-data:
    steps:
      - uses: mqtt/publish
        with:
          concurrency: 8
          topic: factory/{table}/{location}
          qos: 1

查看源码

生成器方式生成数据并发布数据到 Kafka Broker 示例

该示例展示了如何使用 taosgen 工具模拟一万台智能电表，每台智能电表采集电流、电压、相位、位置四个物理量，它们每隔 5 分钟产生一条记录，电流的数据用随机数，电压用正弦波模拟，产生的这些数据发布到 Kafka。

配置详解：

• Kafka 配置参数

  • 连接信息：使用 bootstrap_servers 描述连接 Kafka Broker 的信息。
  • 主题配置 (topic)：使用主题 factory-electric-meter。

• schema 配置参数

  • 名称：指定 schema 的名称。
  • 表名称：定义生成一万张逻辑表的名称规则，格式为 d0 到 d9999，用于组织和标识生成的数据。
  • 表字段结构信息：定义数据表结构，包含 4 个普通列（电流、电压、相位、设备位置）。
    • 时间戳：配置了时间戳生成策略，从指定时间戳 1700000000000 (2023-11-14 22:13:20 UTC) 开始，以 5 分钟的步长递增。
    • 时序数据：current、phase 和 location 使用指定范围的随机数，voltage 使用正弦波模拟。
  • 数据生成行为：使用交错模式写入，每张表写入 1 万条记录，每批写入请求最大行数为 1 万行。

• 数据发布：使用 8 线程并发向 Kafka Broker 发布数据，提高吞吐量。

  • acks：消息确认等级，此示例设为 '1'，表示仅需 Leader 确认。

场景说明：

此配置专为向 Kafka 消息代理发布模拟设备数据而设计。它适用于以下场景：

• Kafka 生产者性能压测： 模拟大规模设备并发写入场景，测试 Kafka Broker 的吞吐能力、网络带宽占用及生产者端的资源消耗，验证不同压缩算法（如 gzip, zstd）对性能的影响。
• 流处理系统集成测试： 向 Kafka 发布结构化的设备数据流，用于测试基于 Flink、Spark Streaming、ksqlDB 或 Pulsar Functions 等流处理引擎的数据接入、窗口计算、状态管理与实时告警功能。
• 物联网平台数据接入验证： 快速构建一个高并发的设备数据注入环境，模拟万台智能电表上报数据，验证 IoT 平台后端服务从 Kafka 消费数据、解析、入库（如时序数据库 TDengine）的完整链路稳定性。
• 规则引擎与消息路由测试： 利用 Kafka 主题的分层命名（如 factory-electric-meter）和消息 Key（如 {table} 表示设备 ID），测试基于 Kafka Connect 或自定义消费者组的消息过滤、多路复用、按设备标签（location）进行动态路由的能力。
• 实时数据管道压力测试： 模拟持续高频率的时序数据流（每 5 分钟/条 × 10,000 台设备），评估从 Kafka 到下游系统（如数据湖、数仓、监控面板）的端到端延迟、积压情况和消费速率匹配度。
• 安全认证机制验证： 配置 SASL_SSL 和 SCRAM-SHA-256 认证，用于测试启用了身份验证和加密传输的 Kafka 集群在真实生产环境下的客户端连接稳定性与安全性。
• 数据格式兼容性测试： 使用 JSON 或 InfluxDB Line Protocol 格式序列化消息体，验证下游消费者或中间件对不同数据格式的解析能力，确保协议兼容性和字段映射正确性。
• 灾备与高可用演练： 在多 Broker 集群环境下，通过高并发写入测试 Kafka 的副本同步、Leader 选举、Broker 故障转移等高可用机制的表现，确保数据不丢失、服务不间断。

kafka:
  bootstrap_servers: localhost:9092
  topic: factory-electric-meter

schema:
  name: meters
  tbname:
    prefix: d
    count: 10000
    from: 0
  columns:
    - name: ts
      type: timestamp
      start: 1700000000000
      precision : ms
      step: 300s
    - name: current
      type: float
      min: 0
      max: 100
    - name: voltage
      type: int
      expr: '220 * math.sqrt(2) * math.sin(_i)'
    - name: phase
      type: float
      min: 0
      max: 360
    - name: location
      type: varchar(20)
      values:
        - Chicago
        - Houston
        - Phoenix
        - Philadelphia
        - Dallas
        - Austin
  generation:
    interlace: 1
    concurrency: 8
    rows_per_table: 10000
    rows_per_batch: 10000
    num_cached_batches: 0

jobs:
  # Kafka produce job
  produce-data:
    steps:
      - uses: kafka/produce
        with:
          concurrency: 8
          acks: 1

查看源码

生成器方式生成数据并写入 InfluxDB 示例

该示例展示了如何使用 taosgen 工具模拟 10 台服务器的 CPU 监控数据，每台服务器每隔 10 秒产生一条包含 CPU 使用率指标的记录，产生的数据通过 InfluxDB v2 Write API 以行协议格式写入 InfluxDB。

配置详解：

• InfluxDB 配置参数
  • 连接信息：通过 url 指定 InfluxDB 服务地址，通过 token 进行身份验证。
  • 目标：将数据写入指定组织的 Bucket 中。
• schema 配置参数
  • 名称：指定 measurement 名称为 cpu，与 Telegraf 的 cpu 采集插件输出的 measurement 名称一致。
  • 表名称：定义生成 10 张逻辑表的名称规则，格式为 host_0 到 host_9，用于组织和标识生成的数据。
  • 表字段结构信息：定义 4 个 field（usage_idle、usage_system、usage_user、usage_iowait）和 2 个 tag（host、cpu）。
    • 时间戳：使用纳秒精度，以 10 秒为步长递增，模拟 Telegraf 默认的 10 秒采集间隔。
    • 时序数据：各 CPU 指标使用指定范围的随机数模拟。
    • 标签数据：host 标签使用 10 个服务器名称，cpu 标签使用 5 个 CPU 核心标识。
  • 数据生成行为：使用交错模式，每张表 100 条记录，每批 500 行。
• 数据写入：使用 2 线程并发以行协议格式写入 InfluxDB，启用 gzip 压缩以减少带宽占用。

场景说明：

此配置专为模拟 Telegraf 采集的系统监控数据写入 InfluxDB 而设计。它适用于以下场景：

• InfluxDB 写入性能测试：模拟大规模服务器集群的监控数据写入，测试 InfluxDB 在高并发行协议写入下的吞吐量和延迟表现。
• Telegraf 数据模拟：生成与 Telegraf cpu 插件输出格式完全一致的数据，用于在无真实设备的环境中测试数据管道。
• InfluxDB 到 TDengine 迁移验证：先向 InfluxDB 写入模拟数据，再测试从 InfluxDB 迁移数据到 TDengine 的完整流程。
• 监控平台集成测试：为 Grafana 等监控平台提供持续的模拟数据流，用于验证仪表盘配置和告警规则。

influxdb:
  url: http://localhost:8086
  token: "your-influxdb-api-token"
  org: default
  bucket: default

schema:
  name: cpu
  tbname:
    prefix: host_
    count: 10
  columns:
    - name: ts
      type: timestamp
      start: now
      precision: ns
      step: 10s
    - name: usage_idle
      type: float
      min: 50
      max: 99
    - name: usage_system
      type: float
      min: 0
      max: 20
    - name: usage_user
      type: float
      min: 0
      max: 50
    - name: usage_iowait
      type: float
      min: 0
      max: 10
  tags:
    - name: host
      type: binary(64)
      values: [server01, server02, server03, server04, server05, server06, server07, server08, server09, server10]
    - name: cpu
      type: binary(16)
      values: [cpu-total, cpu0, cpu1, cpu2, cpu3]
  generation:
    interlace: 1
    rows_per_table: 100
    rows_per_batch: 500

jobs:
  write-data:
    steps:
      - uses: influxdb/write
        with:
          concurrency: 2
          precision: ns
          batch_size: 500
          gzip: true

查看源码