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 执行
);