环境信息：
IoTDB 版本：2.0.8
Java：8
部署方式：Standalone
Sink：write-back-sink
自定义插件类型：PipeProcessor
希望基于 IoTDB Pipe Framework 实现一个对插入值范围计算告警等级的功能。
源数据路径：
root.s1_5w.sub_[0-49999]
测点：
v1
v2
v3
v4
数据格式：
timestamp
ins_ts
v1
v2
v3
v4
告警规则：
value < 80 -> -2
80 <= value < 90 -> -1
90 <= value < 110 -> 0
110 <= value < 120 -> 1
value >= 120 -> 2
例如：
device = sub_0
v1 = 190
v2 = 95
v3 = 70
v4 = 115
期望生成：
field_name=v1, device_name=sub_0, val=190, alarm_level=2
field_name=v3, device_name=sub_0, val=70, alarm_level=-2
field_name=v4, device_name=sub_0, val=115, alarm_level=1
并写回：
root.alarm_s1_5w.alarm_sub_0_v1
root.alarm_s1_5w.alarm_sub_0_v3
root.alarm_s1_5w.alarm_sub_0_v4
问题1：TsFileInsertionEvent 超时
当实现：
@OverRide
public void process(
TsFileInsertionEvent tsFileInsertionEvent,
EventCollector eventCollector) throws Exception {
for (TabletInsertionEvent tabletInsertionEvent :
tsFileInsertionEvent.toTabletInsertionEvents()) {
process(tabletInsertionEvent, eventCollector);
}
}
时，经常出现：
PipeTsFileInsertionEvent.toTabletInsertionEvents()
问题2：忽略 TsFileInsertionEvent 后无输出
将代码改为：
@OverRide
public void process(
TsFileInsertionEvent event,
EventCollector collector) {
// ignore
}
之后超时消失。
但 Processor 没有生成任何告警数据。
日志中也看不到：
AlarmProcessor process tablet
问题3：Pipe 正常运行但没有写回数据
SHOW PIPES：
State = RUNNING
ExceptionMessage = 空
RemainingEventCount = 0
Pipe 没有异常。
Processor 中执行：
eventCollector.collect(
new PipeRawTabletInsertionEvent(alarmTablet, false)
);
后也没有报错。
但是：
SHOW TIMESERIES root.alarm_s1_5w.**
返回为空。
问题：
IoTDB 2.0.8 中，自定义 Processor 生成新 Tablet 的正确方式是什么？
PipeRawTabletInsertionEvent(alarmTablet, false) 是否足够？
是否需要额外的 ProgressIndex、EnrichedEvent 或其他上下文信息？
write-back-sink 是否支持自定义 Processor 动态生成的 Tablet 回写？
package com.xlkh.tsdb.iotdb.plugin.pipe;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.List;
import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.pipe.api.PipeProcessor;
import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.collector.EventCollector;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeProcessorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.BitMap;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@TreeModel
public class AlarmProcessor implements PipeProcessor {
private static final Logger logger = LoggerFactory.getLogger(AlarmProcessor.class);
private static final String DEFAULT_FIELDS = "v1,v2,v3,v4";
private static final String DEFAULT_TARGET_PREFIX = "root.alarm_s1_5w";
private String[] fields;
private String targetPrefix;
private final List<IMeasurementSchema> alarmSchemas = Arrays.asList(
new MeasurementSchema("field_name", TSDataType.TEXT),
new MeasurementSchema("device_name", TSDataType.TEXT),
new MeasurementSchema("val", TSDataType.DOUBLE),
new MeasurementSchema("alarm_level", TSDataType.INT32),
new MeasurementSchema("ts", TSDataType.TIMESTAMP),
new MeasurementSchema("ins_ts", TSDataType.TIMESTAMP)
);
@Override
public void validate(PipeParameterValidator validator) {
}
@Override
public void customize(
PipeParameters parameters, PipeProcessorRuntimeConfiguration configuration) {
this.fields =
parameters
.getStringOrDefault("fields", DEFAULT_FIELDS)
.replace(" ", "")
.split(",");
this.targetPrefix =
parameters.getStringOrDefault("target-prefix", DEFAULT_TARGET_PREFIX);
}
@Override
public void process(
TabletInsertionEvent tabletInsertionEvent,
EventCollector eventCollector) throws Exception {
tabletInsertionEvent.processTablet(
(tablet, rowCollector) -> {
try {
processTablet(tablet, eventCollector);
} catch (Exception e) {
throw new RuntimeException(e);
}
});
}
@Override
public void process(Event event, EventCollector eventCollector) throws Exception {
if (event instanceof TabletInsertionEvent) {
process((TabletInsertionEvent) event, eventCollector);
}
}
private void processTablet(Tablet sourceTablet, EventCollector eventCollector)
throws Exception {
final int rowSize = sourceTablet.getRowSize();
if (rowSize <= 0) {
return;
}
final String sourceDevice = sourceTablet.getDeviceId();
final String deviceName = extractDeviceName(sourceDevice);
logger.info("AlarmProcessor process tablet, device={}, rows={}", deviceName, rowSize);
final int insTsIndex = findMeasurementIndex(sourceTablet, "ins_ts");
for (String field : fields) {
final int fieldIndex = findMeasurementIndex(sourceTablet, field);
if (fieldIndex < 0) {
continue;
}
Tablet alarmTablet =
new Tablet(
targetPrefix + ".alarm_" + deviceName + "_" + field,
alarmSchemas,
rowSize);
int alarmRow = 0;
for (int row = 0; row < rowSize; row++) {
if (isNull(sourceTablet, fieldIndex, row)) {
continue;
}
Object rawValue = getValue(sourceTablet, fieldIndex, row);
if (!(rawValue instanceof Number)) {
continue;
}
double value = ((Number) rawValue).doubleValue();
int alarmLevel = alarmLevel(value);
if (alarmLevel == 0) {
continue;
}
long ts = sourceTablet.getTimestamps()[row];
long insTs = ts;
if (insTsIndex >= 0 && !isNull(sourceTablet, insTsIndex, row)) {
Object insTsValue = getValue(sourceTablet, insTsIndex, row);
if (insTsValue instanceof Number) {
insTs = ((Number) insTsValue).longValue();
}
}
alarmTablet.addTimestamp(alarmRow, ts);
alarmTablet.addValue(
"field_name", alarmRow, new Binary(field, StandardCharsets.UTF_8));
alarmTablet.addValue(
"device_name", alarmRow, new Binary(deviceName, StandardCharsets.UTF_8));
alarmTablet.addValue("val", alarmRow, value);
alarmTablet.addValue("alarm_level", alarmRow, alarmLevel);
alarmTablet.addValue("ts", alarmRow, ts);
alarmTablet.addValue("ins_ts", alarmRow, insTs);
alarmRow++;
}
logger.info("AlarmProcessor collect alarm tablet, device={}, rows={}",
alarmTablet.getDeviceId(), alarmRow);
if (alarmRow > 0) {
alarmTablet.setRowSize(alarmRow);
eventCollector.collect(new PipeRawTabletInsertionEvent(alarmTablet, false));
}
}
}
private int alarmLevel(double value) {
if (value < 80) {
return -2;
}
if (value < 90) {
return -1;
}
if (value < 110) {
return 0;
}
if (value < 120) {
return 1;
}
return 2;
}
private int findMeasurementIndex(Tablet tablet, String measurement) {
List<IMeasurementSchema> schemas = tablet.getSchemas();
for (int i = 0; i < schemas.size(); i++) {
if (measurement.equals(schemas.get(i).getMeasurementName())) {
return i;
}
}
return -1;
}
private boolean isNull(Tablet tablet, int columnIndex, int rowIndex) {
BitMap[] bitMaps = tablet.getBitMaps();
return bitMaps != null
&& bitMaps[columnIndex] != null
&& bitMaps[columnIndex].isMarked(rowIndex);
}
private Object getValue(Tablet tablet, int columnIndex, int rowIndex) {
Object column = tablet.getValues()[columnIndex];
if (column instanceof int[]) {
return ((int[]) column)[rowIndex];
}
if (column instanceof long[]) {
return ((long[]) column)[rowIndex];
}
if (column instanceof float[]) {
return ((float[]) column)[rowIndex];
}
if (column instanceof double[]) {
return ((double[]) column)[rowIndex];
}
if (column instanceof Binary[]) {
return ((Binary[]) column)[rowIndex];
}
if (column instanceof Object[]) {
return ((Object[]) column)[rowIndex];
}
return null;
}
@Override
public void process(
TsFileInsertionEvent tsFileInsertionEvent,
EventCollector eventCollector) {
logger.info("AlarmProcessor ignore TsFileInsertionEvent");
}
private String extractDeviceName(String devicePath) {
int index = devicePath.lastIndexOf('.');
return index >= 0 ? devicePath.substring(index + 1) : devicePath;
}
@Override
public void close() {
}
}CREATE
PIPEPLUGIN alarm_processor AS 'com.xlkh.tsdb.iotdb.plugin.pipe.AlarmProcessor'
USING URI 'file:///home/iotdb/apache-iotdb-2.0.8-all-bin/ext/pipe/alarm-processor-ext.jar';
CREATE
PIPE alarm_pipe
WITH SOURCE (
'source' = 'iotdb-source',
'source.path' = 'root.s1_5w.**',
'source.inclusion' = 'data.insert',
)
WITH PROCESSOR (
'processor' = 'alarm_processor',
'fields' = 'v1,v2,v3,v4',
'target-prefix' = 'root.alarm_s1_5w'
)
WITH SINK (
'sink' = 'write-back-sink'
);
环境信息：
IoTDB 版本：2.0.8
Java：8
部署方式：Standalone
Sink：write-back-sink
自定义插件类型：PipeProcessor
希望基于 IoTDB Pipe Framework 实现一个对插入值范围计算告警等级的功能。
源数据路径：
root.s1_5w.sub_[0-49999]
测点：
v1
v2
v3
v4
数据格式：
timestamp
ins_ts
v1
v2
v3
v4
告警规则：
value < 80 -> -2
80 <= value < 90 -> -1
90 <= value < 110 -> 0
110 <= value < 120 -> 1
value >= 120 -> 2
例如：
device = sub_0
v1 = 190
v2 = 95
v3 = 70
v4 = 115
期望生成：
field_name=v1, device_name=sub_0, val=190, alarm_level=2
field_name=v3, device_name=sub_0, val=70, alarm_level=-2
field_name=v4, device_name=sub_0, val=115, alarm_level=1
并写回：
root.alarm_s1_5w.alarm_sub_0_v1
root.alarm_s1_5w.alarm_sub_0_v3
root.alarm_s1_5w.alarm_sub_0_v4
问题1：TsFileInsertionEvent 超时
当实现：
@OverRide
public void process(
TsFileInsertionEvent tsFileInsertionEvent,
EventCollector eventCollector) throws Exception {
for (TabletInsertionEvent tabletInsertionEvent :
tsFileInsertionEvent.toTabletInsertionEvents()) {
process(tabletInsertionEvent, eventCollector);
}
}
时，经常出现：
PipeTsFileInsertionEvent.toTabletInsertionEvents()
问题2：忽略 TsFileInsertionEvent 后无输出
将代码改为：
@OverRide
public void process(
TsFileInsertionEvent event,
EventCollector collector) {
// ignore
}
之后超时消失。
但 Processor 没有生成任何告警数据。
日志中也看不到：
AlarmProcessor process tablet
问题3：Pipe 正常运行但没有写回数据
SHOW PIPES：
State = RUNNING
ExceptionMessage = 空
RemainingEventCount = 0
Pipe 没有异常。
Processor 中执行：
eventCollector.collect(
new PipeRawTabletInsertionEvent(alarmTablet, false)
);
后也没有报错。
但是：
SHOW TIMESERIES root.alarm_s1_5w.**
返回为空。
问题：
IoTDB 2.0.8 中，自定义 Processor 生成新 Tablet 的正确方式是什么？
PipeRawTabletInsertionEvent(alarmTablet, false) 是否足够？
是否需要额外的 ProgressIndex、EnrichedEvent 或其他上下文信息？
write-back-sink 是否支持自定义 Processor 动态生成的 Tablet 回写？