Apache IoTDB 触发器实操步骤

CREATE TRIGGER [IF NOT EXISTS] <trigger_name>
ON ( <storage_group_path> | <device_path> | <time_series_path> )
[ WHEN ( <condition_expression> ) ]
AFTER | BEFORE ( INSERT | UPDATE | DELETE )
[ ASYNC ]  -- 可选，默认同步执行
DO <sql_action>;

-- 创建存储组
CREATE STORAGE GROUP IF NOT EXISTS /root/sg1;
-- 创建设备 dev1 的温度、湿度序列
CREATE TIMESERIES IF NOT EXISTS /root/sg1/dev1/temperature WITH DATATYPE=FLOAT, ENCODING=PLAIN;
CREATE TIMESERIES IF NOT EXISTS /root/sg1/dev1/humidity WITH DATATYPE=FLOAT, ENCODING=PLAIN;
-- 创建告警序列（存储告警信息）
CREATE TIMESERIES IF NOT EXISTS /root/sg1/dev1/alarm WITH DATATYPE=TEXT, ENCODING=PLAIN;

CREATE TRIGGER IF NOT EXISTS temp_alarm_trigger
ON /root/sg1/dev1  -- 设备级触发，该设备下任意序列插入数据都会检查条件
WHEN (temperature > 80.0)  -- 仅温度超过 80℃ 时触发
AFTER INSERT  -- 数据插入后执行
ASYNC  -- 异步执行，不影响温度数据写入
DO INSERT INTO /root/sg1/dev1 (alarm) VALUES (NOW(), '温度超标告警：' || CAST(temperature AS TEXT) || '℃');

-- 插入正常温度数据（不会触发告警）
INSERT INTO /root/sg1/dev1 (temperature, humidity) VALUES (75.5, 40.0);
-- 插入超标温度数据（触发告警）
INSERT INTO /root/sg1/dev1 (temperature, humidity) VALUES (85.0, 35.0);
-- 查询告警序列，验证结果
SELECT alarm FROM /root/sg1/dev1 WHERE time >= NOW() - 10s;

CREATE TRIGGER IF NOT EXISTS temp_validate_trigger
ON /root/sg1/dev1/temperature  -- 仅针对温度序列触发
WHEN (temperature < -40.0 OR temperature > 120.0)  -- 非法数据条件
BEFORE INSERT  -- 插入前执行
DO REJECT;  -- 内置动作：拒绝插入该条数据

-- 插入 -50℃（非法），不会写入成功
INSERT INTO /root/sg1/dev1 (temperature) VALUES (-50.0);
-- 查询温度序列，无该条数据
SELECT temperature FROM /root/sg1/dev1;

SHOW TRIGGERS;
-- 或查询指定范围的触发器
SHOW TRIGGERS ON /root/sg1/dev1;

DESCRIBE TRIGGER <trigger_name>;

ALTER TRIGGER <trigger_name>
[ WHEN ( <new_condition> ) ]
[ DO <new_sql_action> ]
[ ASYNC | SYNC ];  -- 修改执行模式

DROP TRIGGER IF EXISTS <trigger_name>;

<dependency>
         <groupId>org.apache.iotdb</groupId>
         <artifactId>iotdb-server</artifactId>
         <version>1.2.2</version>  <!-- 与 IoTDB 服务器版本一致 -->
         <scope>provided</scope>  <!-- 服务器已包含该依赖，无需打包 -->
</dependency>
<dependency>
         <groupId>org.apache.iotdb</groupId>
         <artifactId>iotdb-jdbc</artifactId>
         <version>1.2.2</version>
</dependency>

import org.apache.iotdb.db.engine.trigger.api.Trigger;
import org.apache.iotdb.db.engine.trigger.api.TriggerContext;
import org.apache.iotdb.db.engine.trigger.api.TriggerEvent;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.write.record.Tablet;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
public class KafkaSyncTrigger implements Trigger {
         private KafkaProducer<String, String> kafkaProducer;
         private String topic;
         // 初始化：创建 Kafka 连接（配置从触发器参数传入）
         @Override
         public void init(TriggerContext context) throws Exception {
             // 从触发器创建时的参数中获取 Kafka 配置
             String bootstrapServers = context.getTriggerAttributes().get("bootstrap.servers");
             this.topic = context.getTriggerAttributes().get("topic");
             Properties props = new Properties();
             props.put("bootstrap.servers", bootstrapServers);
             props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
             props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
             this.kafkaProducer = new KafkaProducer<>(props);
         }
         // 插入数据时触发：将数据同步到 Kafka
         @Override
         public void onInsert(TriggerEvent event) throws Exception {
             Tablet tablet = event.getTablet();  // 获取插入的数据（Tablet 是 IoTDB 批量数据存储结构）
             String device = tablet.getDeviceId();  // 设备 ID
             // 遍历 Tablet 中的数据行，构造 Kafka 消息
             for (int i = 0; i < tablet.getRowSize(); i++) {
                 long timestamp = tablet.getTimestamps()[i];  // 时间戳
                 String temperature = tablet.getValues()[0][i].toString();  // 假设第一个序列是温度
                 String message = String.format("device:%s,time:%d,temperature:%s", device, timestamp, temperature);
                 // 发送到 Kafka
                 kafkaProducer.send(new ProducerRecord<>(topic, device, message));
             }
         }
         // 销毁：关闭 Kafka 连接
         @Override
         public void close() throws Exception {
             if (kafkaProducer != null) {
                 kafkaProducer.close();
             }
         }
         // 无需处理更新和删除事件，可空实现
         @Override
         public void onUpdate(TriggerEvent event) throws Exception {}
         @Override
         public void onDelete(TriggerEvent event) throws Exception {}
}

CREATE TRIGGER IF NOT EXISTS kafka_sync_trigger
ON /root/sg1/dev1  -- 设备级触发
AFTER INSERT
ASYNC  -- 异步执行，避免影响写入
WITH (
         'trigger.class' = 'com.example.KafkaSyncTrigger',  -- 自定义类的全限定名
         'bootstrap.servers' = '192.168.1.100:9092',  -- Kafka 地址（自定义参数）
         'topic' = 'iotdb_data_sync'  -- Kafka 主题（自定义参数）
);

-- 插入数据
INSERT INTO /root/sg1/dev1 (temperature) VALUES (78.5);
-- 查看 Kafka 主题 iotdb_data_sync，应收到消息：device:/root/sg1/dev1,time:xxx,temperature:78.5

// 在 onInsert 中调用 UDF 函数计算温度平均值
UDFExecutor executor = new UDFExecutor("avg_temp_udf");  // avg_temp_udf 是已注册的 UDF
Object result = executor.calculate(tablet.getValues()[0]);  // 传入温度数据

-- 先创建连续查询
CREATE CONTINUOUS QUERY cq_avg_temp
BEGIN
         INSERT INTO /root/sg1/dev1/avg_temp SELECT AVG(temperature) FROM /root/sg1/dev1 GROUP BY TIME(1m)
END;
-- 创建触发器，监听 avg_temp 序列的插入（CQ 执行结果）
CREATE TRIGGER cq_sync_trigger
ON /root/sg1/dev1/avg_temp
AFTER INSERT
ASYNC
DO INSERT INTO /root/sg1/dashboard/avg_temp_sync VALUES (NOW(), avg_temp);

trigger.async.thread.pool.size=20  # 调整为合适的线程数

// Kafka 发送超时控制
kafkaProducer.send(record).get(5, TimeUnit.SECONDS);  // 5 秒超时

CREATE TRIGGER cluster_trigger
ON /root/sg1
AFTER INSERT
WITH (
         'trigger.class' = 'com.example.ClusterTrigger',
         'NODE_LIST' = 'node1,node2'  -- 仅在 node1 和 node2 执行
);