编写触发器

触发器依赖

触发器的逻辑需要您编写 Java 类进行实现。 在编写触发器逻辑时，需要使用到下面展示的依赖。如果您使用 Mavenopen in new window，则可以直接从 Maven 库open in new window中搜索到它们。请注意选择和目标服务器版本相同的依赖版本。

1. <dependency>
2. <groupId>org.apache.iotdb</groupId>
3. <artifactId>iotdb-server</artifactId>
4. <version>1.0.0</version>
5. <scope>provided</scope>
6. </dependency>

接口说明

编写一个触发器需要实现 org.apache.iotdb.trigger.api.Trigger 类。

1. import org.apache.iotdb.trigger.api.enums.FailureStrategy;
2. import org.apache.iotdb.tsfile.write.record.Tablet;

4. public interface Trigger {

6. /**
7. * This method is mainly used to validate {@link TriggerAttributes} before calling {@link
8. * Trigger#onCreate(TriggerAttributes)}.
9. *
10. * @param attributes TriggerAttributes
11. * @throws Exception e
12. */
13. default void validate(TriggerAttributes attributes) throws Exception {}

15. /**
16. * This method will be called when creating a trigger after validation.
17. *
18. * @param attributes TriggerAttributes
19. * @throws Exception e
20. */
21. default void onCreate(TriggerAttributes attributes) throws Exception {}

23. /**
24. * This method will be called when dropping a trigger.
25. *
26. * @throws Exception e
27. */
28. default void onDrop() throws Exception {}

30. /**
31. * When restarting a DataNode, Triggers that have been registered will be restored and this method
32. * will be called during the process of restoring.
33. *
34. * @throws Exception e
35. */
36. default void restore() throws Exception {}

38. /**
39. * Overrides this method to set the expected FailureStrategy, {@link FailureStrategy#OPTIMISTIC}
40. * is the default strategy.
41. *
42. * @return {@link FailureStrategy}
43. */
44. default FailureStrategy getFailureStrategy() {
45. return FailureStrategy.OPTIMISTIC;
46. }

48. /**
49. * @param tablet see {@link Tablet} for detailed information of data structure. Data that is
50. *     inserted will be constructed as a Tablet and you can define process logic with {@link
51. *     Tablet}.
52. * @return true if successfully fired
53. * @throws Exception e
54. */
55. default boolean fire(Tablet tablet) throws Exception {
56. return true;
57. }
58. }

该类主要提供了两类编程接口：生命周期相关接口和数据变动侦听相关接口。该类中所有的接口都不是必须实现的，当您不实现它们时，它们不会对流经的数据操作产生任何响应。您可以根据实际需要，只实现其中若干接口。

下面是所有可供用户进行实现的接口的说明。

生命周期相关接口

数据变动侦听相关接口

侦听接口

1. /**
2. * @param tablet see {@link Tablet} for detailed information of data structure. Data that is
3. *     inserted will be constructed as a Tablet and you can define process logic with {@link
4. *     Tablet}.
5. * @return true if successfully fired
6. * @throws Exception e
7. */
8. default boolean fire(Tablet tablet) throws Exception {
9. return true;
10. }

数据变动时，触发器以 Tablet 作为触发操作的单位。您可以通过 Tablet 获取相应序列的元数据和数据，然后进行相应的触发操作，触发成功则返回值应当为 true。该接口返回 false 或是抛出异常我们均认为触发失败。在触发失败时，我们会根据侦听策略接口进行相应的操作。

进行一次 INSERT 操作时，对于其中的每条时间序列，我们会检测是否有侦听该路径模式的触发器，然后将符合同一个触发器所侦听的路径模式的时间序列数据组装成一个新的 Tablet 用于触发器的 fire 接口。可以理解成：

1. Map<PartialPath, List<Trigger>> pathToTriggerListMap => Map<Trigger, Tablet>

请注意，目前我们不对触发器的触发顺序有任何保证。

下面是示例：

假设有三个触发器，触发器的触发时机均为 BEFORE INSERT

• 触发器 Trigger1 侦听路径模式：root.sgopen in new window.*
• 触发器 Trigger2 侦听路径模式：root.sg.a
• 触发器 Trigger3 侦听路径模式：root.sg.b

写入语句：

1. insert into root.sg(time, a, b) values (1, 1, 1);

序列 root.sg.a 匹配 Trigger1 和 Trigger2，序列 root.sg.b 匹配 Trigger1 和 Trigger3，那么：

• root.sg.a 和 root.sg.b 的数据会被组装成一个新的 tablet1，在相应的触发时机进行 Trigger1.fire(tablet1)
• root.sg.a 的数据会被组装成一个新的 tablet2，在相应的触发时机进行 Trigger2.fire(tablet2)
• root.sg.b 的数据会被组装成一个新的 tablet3，在相应的触发时机进行 Trigger3.fire(tablet3)

侦听策略接口

在触发器触发失败时，我们会根据侦听策略接口设置的策略进行相应的操作，您可以通过下述接口设置 org.apache.iotdb.trigger.api.enums.FailureStrategy，目前有乐观和悲观两种策略：

• 乐观策略：触发失败的触发器不影响后续触发器的触发，也不影响写入流程，即我们不对触发失败涉及的序列做额外处理，仅打日志记录失败，最后返回用户写入数据成功，但触发部分失败。
• 悲观策略：失败触发器影响后续所有 Pipeline 的处理，即我们认为该 Trigger 触发失败会导致后续所有触发流程不再进行。如果该触发器的触发时机为 BEFORE INSERT，那么写入也不再进行，直接返回写入失败。

1. /**
2. * Overrides this method to set the expected FailureStrategy, {@link FailureStrategy#OPTIMISTIC}
3. * is the default strategy.
4. *
5. * @return {@link FailureStrategy}
6. */
7. default FailureStrategy getFailureStrategy() {
8. return FailureStrategy.OPTIMISTIC;
9. }

您可以参考下图辅助理解，其中 Trigger1 配置采用乐观策略，Trigger2 配置采用悲观策略。Trigger1 和 Trigger2 的触发时机是 BEFORE INSERT，Trigger3 和 Trigger4 的触发时机是 AFTER INSERT。 正常执行流程如下：

示例

如果您使用 Mavenopen in new window，可以参考我们编写的示例项目 trigger-example。您可以在 这里open in new window 找到它。后续我们会加入更多的示例项目供您参考。

下面是其中一个示例项目的代码：

1. /*
2. * Licensed to the Apache Software Foundation (ASF) under one
3. * or more contributor license agreements.  See the NOTICE file
4. * distributed with this work for additional information
5. * regarding copyright ownership.  The ASF licenses this file
6. * to you under the Apache License, Version 2.0 (the
7. * "License"); you may not use this file except in compliance
8. * with the License.  You may obtain a copy of the License at
9. *
10. *     http://www.apache.org/licenses/LICENSE-2.0
11. *
12. * Unless required by applicable law or agreed to in writing,
13. * software distributed under the License is distributed on an
14. * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15. * KIND, either express or implied.  See the License for the
16. * specific language governing permissions and limitations
17. * under the License.
18. */

20. package org.apache.iotdb.trigger;

22. import org.apache.iotdb.db.engine.trigger.sink.alertmanager.AlertManagerConfiguration;
23. import org.apache.iotdb.db.engine.trigger.sink.alertmanager.AlertManagerEvent;
24. import org.apache.iotdb.db.engine.trigger.sink.alertmanager.AlertManagerHandler;
25. import org.apache.iotdb.trigger.api.Trigger;
26. import org.apache.iotdb.trigger.api.TriggerAttributes;
27. import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
28. import org.apache.iotdb.tsfile.write.record.Tablet;
29. import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;

31. import org.slf4j.Logger;
32. import org.slf4j.LoggerFactory;

34. import java.io.IOException;
35. import java.util.HashMap;
36. import java.util.List;

38. public class ClusterAlertingExample implements Trigger {
39. private static final Logger LOGGER = LoggerFactory.getLogger(ClusterAlertingExample.class);

41. private final AlertManagerHandler alertManagerHandler = new AlertManagerHandler();

43. private final AlertManagerConfiguration alertManagerConfiguration =
44. new AlertManagerConfiguration("http://127.0.0.1:9093/api/v2/alerts");

46. private String alertname;

48. private final HashMap<String, String> labels = new HashMap<>();

50. private final HashMap<String, String> annotations = new HashMap<>();

52. @Override
53. public void onCreate(TriggerAttributes attributes) throws Exception {
54. alertname = "alert_test";

56. labels.put("series", "root.ln.wf01.wt01.temperature");
57. labels.put("value", "");
58. labels.put("severity", "");

60. annotations.put("summary", "high temperature");
61. annotations.put("description", "{{.alertname}}: {{.series}} is {{.value}}");

63. alertManagerHandler.open(alertManagerConfiguration);
64. }

66. @Override
67. public void onDrop() throws IOException {
68. alertManagerHandler.close();
69. }

71. @Override
72. public boolean fire(Tablet tablet) throws Exception {
73. List<MeasurementSchema> measurementSchemaList = tablet.getSchemas();
74. for (int i = 0, n = measurementSchemaList.size(); i < n; i++) {
75. if (measurementSchemaList.get(i).getType().equals(TSDataType.DOUBLE)) {
76. // for example, we only deal with the columns of Double type
77. double[] values = (double[]) tablet.values[i];
78. for (double value : values) {
79. if (value > 100.0) {
80. LOGGER.info("trigger value > 100");
81. labels.put("value", String.valueOf(value));
82. labels.put("severity", "critical");
83. AlertManagerEvent alertManagerEvent =
84. new AlertManagerEvent(alertname, labels, annotations);
85. alertManagerHandler.onEvent(alertManagerEvent);
86. } else if (value > 50.0) {
87. LOGGER.info("trigger value > 50");
88. labels.put("value", String.valueOf(value));
89. labels.put("severity", "warning");
90. AlertManagerEvent alertManagerEvent =
91. new AlertManagerEvent(alertname, labels, annotations);
92. alertManagerHandler.onEvent(alertManagerEvent);
93. }
94. }
95. }
96. }
97. return true;
98. }
99. }