coniferous-cmd opened a new issue, #17927:
URL: https://github.com/apache/iotdb/issues/17927
环境信息：
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 回写？
```java
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() {
}
}
```
```sql
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'
);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
