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 ]  -- 可選，默認同步執(zhí)行
DO <sql_action>;

-- 創(chuàng)建存儲組
CREATE STORAGE GROUP IF NOT EXISTS /root/sg1;
-- 創(chuàng)建設(shè)備 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;
-- 創(chuàng)建告警序列（存儲告警信息）
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  -- 設(shè)備級觸發(fā)，該設(shè)備下任意序列插入數(shù)據(jù)都會檢查條件
WHEN (temperature > 80.0)  -- 僅溫度超過 80℃ 時觸發(fā)
AFTER INSERT  -- 數(shù)據(jù)插入后執(zhí)行
ASYNC  -- 異步執(zhí)行，不影響溫度數(shù)據(jù)寫入
DO INSERT INTO /root/sg1/dev1 (alarm) VALUES (NOW(), '溫度超標告警：' || CAST(temperature AS TEXT) || '℃');

-- 插入正常溫度數(shù)據(jù)（不會觸發(fā)告警）
INSERT INTO /root/sg1/dev1 (temperature, humidity) VALUES (75.5, 40.0);
-- 插入超標溫度數(shù)據(jù)（觸發(fā)告警）
INSERT INTO /root/sg1/dev1 (temperature, humidity) VALUES (85.0, 35.0);
-- 查詢告警序列，驗證結(jié)果
SELECT alarm FROM /root/sg1/dev1 WHERE time >= NOW() - 10s;

CREATE TRIGGER IF NOT EXISTS temp_validate_trigger
ON /root/sg1/dev1/temperature  -- 僅針對溫度序列觸發(fā)
WHEN (temperature < -40.0 OR temperature > 120.0)  -- 非法數(shù)據(jù)條件
BEFORE INSERT  -- 插入前執(zhí)行
DO REJECT;  -- 內(nèi)置動作：拒絕插入該條數(shù)據(jù)

-- 插入 -50℃（非法），不會寫入成功
INSERT INTO /root/sg1/dev1 (temperature) VALUES (-50.0);
-- 查詢溫度序列，無該條數(shù)據(jù)
SELECT temperature FROM /root/sg1/dev1;

SHOW TRIGGERS;
-- 或查詢指定范圍的觸發(fā)器
SHOW TRIGGERS ON /root/sg1/dev1;

DESCRIBE TRIGGER <trigger_name>;

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

DROP TRIGGER IF EXISTS <trigger_name>;

<dependency>
         <groupId>org.apache.iotdb</groupId>
         <artifactId>iotdb-server</artifactId>
         <version>1.2.2</version>  <!-- 與 IoTDB 服務(wù)器版本一致 -->
         <scope>provided</scope>  <!-- 服務(wù)器已包含該依賴，無需打包 -->
</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;
         // 初始化：創(chuàng)建 Kafka 連接（配置從觸發(fā)器參數(shù)傳入）
         @Override
         public void init(TriggerContext context) throws Exception {
             // 從觸發(fā)器創(chuàng)建時的參數(shù)中獲取 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);
         }
         // 插入數(shù)據(jù)時觸發(fā)：將數(shù)據(jù)同步到 Kafka
         @Override
         public void onInsert(TriggerEvent event) throws Exception {
             Tablet tablet = event.getTablet();  // 獲取插入的數(shù)據(jù)（Tablet 是 IoTDB 批量數(shù)據(jù)存儲結(jié)構(gòu)）
             String device = tablet.getDeviceId();  // 設(shè)備 ID
             // 遍歷 Tablet 中的數(shù)據(jù)行，構(gòu)造 Kafka 消息
             for (int i = 0; i < tablet.getRowSize(); i++) {
                 long timestamp = tablet.getTimestamps()[i];  // 時間戳
                 String temperature = tablet.getValues()[0][i].toString();  // 假設(shè)第一個序列是溫度
                 String message = String.format("device:%s,time:%d,temperature:%s", device, timestamp, temperature);
                 // 發(fā)送到 Kafka
                 kafkaProducer.send(new ProducerRecord<>(topic, device, message));
             }
         }
         // 銷毀：關(guān)閉 Kafka 連接
         @Override
         public void close() throws Exception {
             if (kafkaProducer != null) {
                 kafkaProducer.close();
             }
         }
         // 無需處理更新和刪除事件，可空實現(xiàn)
         @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  -- 設(shè)備級觸發(fā)
AFTER INSERT
ASYNC  -- 異步執(zhí)行，避免影響寫入
WITH (
         'trigger.class' = 'com.example.KafkaSyncTrigger',  -- 自定義類的全限定名
         'bootstrap.servers' = '192.168.1.100:9092',  -- Kafka 地址（自定義參數(shù)）
         'topic' = 'iotdb_data_sync'  -- Kafka 主題（自定義參數(shù)）
);

-- 插入數(shù)據(jù)
INSERT INTO /root/sg1/dev1 (temperature) VALUES (78.5);
-- 查看 Kafka 主題 iotdb_data_sync，應(yīng)收到消息：device:/root/sg1/dev1,time:xxx,temperature:78.5

// 在 onInsert 中調(diào)用 UDF 函數(shù)計算溫度平均值
UDFExecutor executor = new UDFExecutor("avg_temp_udf");  // avg_temp_udf 是已注冊的 UDF
Object result = executor.calculate(tablet.getValues()[0]);  // 傳入溫度數(shù)據(jù)

-- 先創(chuàng)建連續(xù)查詢
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;
-- 創(chuàng)建觸發(fā)器，監(jiān)聽 avg_temp 序列的插入（CQ 執(zhí)行結(jié)果）
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  # 調(diào)整為合適的線程數(shù)

// Kafka 發(fā)送超時控制
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 執(zhí)行
);