You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by le...@apache.org on 2025/02/23 17:44:33 UTC
(iotdb) 32/44: minmax
This is an automated email from the ASF dual-hosted git repository.
leirui pushed a commit to branch research/area-visualization
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 30c63392fcff1659ac8979a7858472911598f3e4
Author: Lei Rui <10...@qq.com>
AuthorDate: Tue Jan 28 04:25:35 2025 +0800
minmax
---
server/pom.xml | 12 +-
server/sample_minmax-jar-with-dependencies.jar | Bin 0 -> 40761381 bytes
.../org/apache/iotdb/db/query/eBUG/MinMax.java | 241 +++++++++++++++++++++
.../org/apache/iotdb/db/query/eBUG/sample_FSW.java | 4 -
...ample_bottomUpYdiff.java => sample_MinMax.java} | 33 +--
.../iotdb/db/query/eBUG/sample_bottomUpYdiff.java | 1 -
.../apache/iotdb/db/query/eBUG/sample_eBUG.java | 3 +-
7 files changed, 270 insertions(+), 24 deletions(-)
diff --git a/server/pom.xml b/server/pom.xml
index f918a874eb2..1808c5f6791 100644
--- a/server/pom.xml
+++ b/server/pom.xml
@@ -206,6 +206,12 @@
<artifactId>caffeine</artifactId>
<version>2.9.1</version>
</dependency>
+ <dependency>
+ <groupId>org.datanucleus</groupId>
+ <artifactId>datanucleus-core</artifactId>
+ <version>4.1.17</version>
+ <scope>compile</scope>
+ </dependency>
</dependencies>
<build>
<plugins>
@@ -281,12 +287,14 @@
<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<configuration>
- <finalName>sample_fsw</finalName>
+ <finalName>sample_minmax</finalName>
+ <!-- <finalName>sample_fsw</finalName>-->
<!-- <finalName>sample_eBUG</finalName>-->
<!-- <finalName>sample_BUYdiff</finalName>-->
<archive>
<manifest>
- <mainClass>org.apache.iotdb.db.query.eBUG.sample_FSW</mainClass>
+ <mainClass>org.apache.iotdb.db.query.eBUG.sample_MinMax</mainClass>
+ <!-- <mainClass>org.apache.iotdb.db.query.eBUG.sample_FSW</mainClass>-->
<!-- <mainClass>org.apache.iotdb.db.query.eBUG.sample_eBUG</mainClass>-->
<!-- <mainClass>org.apache.iotdb.db.query.eBUG.sample_bottomUpYdiff</mainClass>-->
</manifest>
diff --git a/server/sample_minmax-jar-with-dependencies.jar b/server/sample_minmax-jar-with-dependencies.jar
new file mode 100644
index 00000000000..cd09c6c3bbf
Binary files /dev/null and b/server/sample_minmax-jar-with-dependencies.jar differ
diff --git a/server/src/main/java/org/apache/iotdb/db/query/eBUG/MinMax.java b/server/src/main/java/org/apache/iotdb/db/query/eBUG/MinMax.java
new file mode 100644
index 00000000000..5ee516f0f9f
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/query/eBUG/MinMax.java
@@ -0,0 +1,241 @@
+package org.apache.iotdb.db.query.eBUG;
+
+import java.io.*;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+
+public class MinMax {
+ public enum fixedBUCKETtype {
+ width,
+ frequency,
+ }
+
+ public static List<Point> reducePoints_equalFrequencyBucket(
+ List<Point> points, int m, boolean debug) {
+ // 不算首尾点，还要(m-2)/2个分桶，每个分桶里采集MinMax两个点
+ int numOfBuckets = (int) ((m - 2) / 2.0);
+
+ // 等宽分桶
+ int frequency = (int) Math.ceil(points.size() * 1.0 / numOfBuckets);
+
+ if (debug) {
+ System.out.println("numOfBuckets=" + numOfBuckets);
+ System.out.println("frequency=" + frequency);
+ }
+
+ List<Point> res = new ArrayList<>();
+ res.add(points.get(0));
+
+ int currentBucketIdx = 0;
+ double currentBucketStartIdx = currentBucketIdx * frequency;
+ double currentBucketEndIdx = currentBucketStartIdx + frequency;
+ double minValue = Double.MAX_VALUE;
+ int minIdx = -1;
+ double maxValue = -Double.MAX_VALUE; // Double.MIN_VALUE is positive so do not use it!!!
+ int maxIdx = -1;
+ for (int i = 0; i < points.size() - 1; i++) { // 注意到倒数第二个点为止，因为如果到全局尾点，由于精度出入会有bug
+ Point p = points.get(i);
+ while (i >= currentBucketEndIdx) {
+ // record the results of the last bucket
+ if (minIdx < maxIdx) {
+ res.add(points.get(minIdx));
+ res.add(points.get(maxIdx));
+ } else {
+ res.add(points.get(maxIdx));
+ res.add(points.get(minIdx));
+ }
+ if (debug) {
+ System.out.println(
+ currentBucketStartIdx
+ + ","
+ + currentBucketEndIdx
+ + ","
+ + currentBucketIdx
+ + ","
+ + res.size());
+ }
+ // find the bucket that holds p, and reset for this new bucket
+ currentBucketIdx++;
+ currentBucketStartIdx = currentBucketIdx * frequency;
+ currentBucketEndIdx = currentBucketStartIdx + frequency;
+ minValue = Double.MAX_VALUE;
+ maxValue = -Double.MAX_VALUE;
+ minIdx = -1;
+ maxIdx = -1;
+ }
+ if (p.y < minValue) {
+ minValue = p.y;
+ minIdx = i;
+ }
+ if (p.y > maxValue) {
+ maxValue = p.y;
+ maxIdx = i;
+ }
+ }
+
+ // record the results of the last bucket
+ if (minIdx < maxIdx) {
+ res.add(points.get(minIdx));
+ res.add(points.get(maxIdx));
+ } else {
+ res.add(points.get(maxIdx));
+ res.add(points.get(minIdx));
+ }
+ if (debug) {
+ System.out.println(
+ currentBucketStartIdx
+ + ","
+ + currentBucketEndIdx
+ + ","
+ + currentBucketIdx
+ + ","
+ + res.size());
+ }
+
+ // record the last point
+ res.add(points.get(points.size() - 1));
+
+ return res;
+ }
+
+ public static List<Point> reducePoints_equalWidthBucket(
+ List<Point> points, int m, boolean debug) {
+ double startTime = points.get(0).x;
+ double endTime = points.get(points.size() - 1).x;
+
+ // 不算首尾点，还要(m-2)/2个分桶，每个分桶里采集MinMax两个点
+ int numOfBuckets = (int) ((m - 2) / 2.0);
+
+ // 等宽分桶
+ double interval = (endTime - startTime) / numOfBuckets;
+
+ if (debug) {
+ System.out.println("numOfBuckets=" + numOfBuckets);
+ System.out.println("width=" + interval);
+ }
+
+ List<Point> res = new ArrayList<>();
+ res.add(points.get(0));
+
+ int currentBucketIdx = 0;
+ double currentBucketStartTime = currentBucketIdx * interval;
+ double currentBucketEndTime = currentBucketStartTime + interval;
+ double minValue = Double.MAX_VALUE;
+ int minIdx = -1;
+ double maxValue = -Double.MAX_VALUE; // Double.MIN_VALUE is positive so do not use it!!!
+ int maxIdx = -1;
+ for (int i = 0; i < points.size() - 1; i++) { // 注意到倒数第二个点为止，因为如果到全局尾点，由于精度出入会有bug
+ Point p = points.get(i);
+ while (p.x >= currentBucketEndTime) {
+ // record the results of the last bucket
+ if (minIdx < maxIdx) {
+ res.add(points.get(minIdx));
+ res.add(points.get(maxIdx));
+ } else {
+ res.add(points.get(maxIdx));
+ res.add(points.get(minIdx));
+ }
+ if (debug) {
+ System.out.println(
+ currentBucketStartTime
+ + ","
+ + currentBucketEndTime
+ + ","
+ + currentBucketIdx
+ + ","
+ + res.size());
+ }
+ // find the bucket that holds p, and reset for this new bucket
+ currentBucketIdx++;
+ currentBucketStartTime = currentBucketIdx * interval;
+ currentBucketEndTime = currentBucketStartTime + interval;
+ minValue = Double.MAX_VALUE;
+ maxValue = -Double.MAX_VALUE;
+ minIdx = -1;
+ maxIdx = -1;
+ }
+ if (p.y < minValue) {
+ minValue = p.y;
+ minIdx = i;
+ }
+ if (p.y > maxValue) {
+ maxValue = p.y;
+ maxIdx = i;
+ }
+ }
+
+ // record the results of the last bucket
+ if (minIdx < maxIdx) {
+ res.add(points.get(minIdx));
+ res.add(points.get(maxIdx));
+ } else {
+ res.add(points.get(maxIdx));
+ res.add(points.get(minIdx));
+ }
+ if (debug) {
+ System.out.println(
+ currentBucketStartTime
+ + ","
+ + currentBucketEndTime
+ + ","
+ + currentBucketIdx
+ + ","
+ + res.size());
+ }
+
+ // record the last point
+ res.add(points.get(points.size() - 1));
+
+ return res;
+ }
+
+ public static void main(String[] args) {
+ Random rand = new Random(10);
+ String input = "D:\\datasets\\regular\\tmp2.csv";
+ boolean hasHeader = true;
+ int timeIdx = 0;
+ int valueIdx = 1;
+ int N = -1;
+ List<Point> points = Tool.readFromFile(input, hasHeader, timeIdx, valueIdx, N);
+ // Polyline polyline = new Polyline();
+ // for (int i = 0; i < N; i += 1) {
+ // double v = rand.nextInt(1000);
+ // polyline.addVertex(new Point(i, v));
+ // }
+ // List<Point> points = polyline.getVertices();
+ try (FileWriter writer = new FileWriter("raw.csv")) {
+ // 写入CSV头部
+ writer.append("x,y,z\n");
+
+ // 写入每个点的数据
+ for (Point point : points) {
+ writer.append(point.x + "," + point.y + "," + point.z + "\n");
+ }
+ System.out.println(points.size() + " Data has been written");
+ } catch (IOException e) {
+ System.out.println("Error writing to CSV file: " + e.getMessage());
+ }
+ int m = 10;
+
+ long startTime = System.currentTimeMillis();
+ List<Point> sampled = reducePoints_equalFrequencyBucket(points, m, false);
+ long endTime = System.currentTimeMillis();
+ System.out.println("Time taken: " + (endTime - startTime) + "ms");
+
+ // for (Point p : sampled) {
+ // System.out.println(p);
+ // }
+ System.out.println(sampled.size());
+
+ try (PrintWriter writer = new PrintWriter(new File("output.csv"))) {
+ // 写入字符串
+ for (int i = 0; i < sampled.size(); i++) {
+ writer.println(sampled.get(i).x + "," + sampled.get(i).y);
+ }
+
+ } catch (FileNotFoundException e) {
+ e.printStackTrace();
+ }
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/eBUG/sample_FSW.java b/server/src/main/java/org/apache/iotdb/db/query/eBUG/sample_FSW.java
index 426c99f4638..098cf1e36ff 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/eBUG/sample_FSW.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/eBUG/sample_FSW.java
@@ -9,9 +9,6 @@ import static org.apache.iotdb.db.query.eBUG.Tool.generateOutputFileName;
import static org.apache.iotdb.db.query.eBUG.Tool.getParam;
public class sample_FSW {
- // 输入一条时间序列 t,v
- // 输出按照bottom-up淘汰顺序排列的dominated significance,t,v。
- // 用于后期在线采样时选取倒数m个点（也就是DS最大的m个点，或者最晚淘汰的m个点）作为采样结果（选出之后要自行把这m个点重新按照时间戳x递增排列）
public static void main(String[] args) throws IOException {
if (args.length < 7) {
System.out.println(
@@ -65,7 +62,6 @@ public class sample_FSW {
+ (endTime - startTime)
+ "ms");
- // 输出结果到csv，按照z,x,y三列，因为results结果已经按照z（即DS）递增排序，对应bottom-up的淘汰顺序，越小代表越早被淘汰
try (BufferedWriter writer = new BufferedWriter(new FileWriter(outputFile))) {
writer.write("x,y");
writer.newLine();
diff --git a/server/src/main/java/org/apache/iotdb/db/query/eBUG/sample_bottomUpYdiff.java b/server/src/main/java/org/apache/iotdb/db/query/eBUG/sample_MinMax.java
similarity index 71%
copy from server/src/main/java/org/apache/iotdb/db/query/eBUG/sample_bottomUpYdiff.java
copy to server/src/main/java/org/apache/iotdb/db/query/eBUG/sample_MinMax.java
index ec96d1d45e8..767b7f7151e 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/eBUG/sample_bottomUpYdiff.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/eBUG/sample_MinMax.java
@@ -7,11 +7,11 @@ import java.util.List;
import static org.apache.iotdb.db.query.eBUG.Tool.generateOutputFileName;
-public class sample_bottomUpYdiff {
+public class sample_MinMax {
public static void main(String[] args) throws IOException {
if (args.length < 7) {
System.out.println(
- "Usage: Please provide arguments: inputFilePath,hasHeader,timeIdx,valueIdx,N,m,errorType,(outDir)");
+ "Usage: Please provide arguments: inputFilePath,hasHeader,timeIdx,valueIdx,N,m,bucketType,(outDir)");
}
String input = args[0];
boolean hasHeader = Boolean.parseBoolean(args[1]);
@@ -19,16 +19,14 @@ public class sample_bottomUpYdiff {
int valueIdx = Integer.parseInt(args[3]);
int N = Integer.parseInt(args[4]); // N<=0表示读全部行，N>0表示读最多N行
int m = Integer.parseInt(args[5]);
- String errorTypeStr = args[6];
- DP.ERRORtype errorType;
- if (errorTypeStr.equals("L1")) {
- errorType = DP.ERRORtype.L1;
- } else if (errorTypeStr.equals("L2")) {
- errorType = DP.ERRORtype.L2;
- } else if (errorTypeStr.equals("L_infy")) {
- errorType = DP.ERRORtype.L_infy;
+ String bucketTypeStr = args[6];
+ MinMax.fixedBUCKETtype bucketType;
+ if (bucketTypeStr.equals("width")) {
+ bucketType = MinMax.fixedBUCKETtype.width;
+ } else if (bucketTypeStr.equals("frequency")) {
+ bucketType = MinMax.fixedBUCKETtype.frequency;
} else {
- throw new IOException("please input errorType as L1/L2/L_infy");
+ throw new IOException("please input fixed bucket type as width/frequency");
}
String outDir;
if (args.length > 7) {
@@ -44,20 +42,24 @@ public class sample_bottomUpYdiff {
System.out.println("Value index: " + valueIdx);
System.out.println("N: " + N);
System.out.println("m: " + m);
- System.out.println("errorType: " + errorType);
+ System.out.println("bucketType: " + bucketType);
System.out.println("outDir: " + outDir);
// 读取原始序列
List<Point> points = Tool.readFromFile(input, hasHeader, timeIdx, valueIdx, N);
String outputFile =
- generateOutputFileName(input, outDir, "-BU" + errorType + "-n" + points.size() + "-m" + m);
+ generateOutputFileName(input, outDir, "-" + bucketType + "-n" + points.size() + "-m" + m);
System.out.println("Output file: " + outputFile); // do not modify this hint string log
// 采样
+ List<Point> results;
long startTime = System.currentTimeMillis();
- // List<Point> results = SWAB.seg_bottomUp_m_withTimestamps(points, m, null, false);
- List<Point> results = BottomUpYdiff.reducePoints(points, m, errorType, false);
+ if (bucketType == MinMax.fixedBUCKETtype.width) {
+ results = MinMax.reducePoints_equalWidthBucket(points, m, false);
+ } else {
+ results = MinMax.reducePoints_equalFrequencyBucket(points, m, false);
+ }
long endTime = System.currentTimeMillis();
System.out.println("result point number: " + results.size());
@@ -70,7 +72,6 @@ public class sample_bottomUpYdiff {
+ (endTime - startTime)
+ "ms");
- // 输出结果到csv，按照z,x,y三列，因为results结果已经按照z（即DS）递增排序，对应bottom-up的淘汰顺序，越小代表越早被淘汰
try (BufferedWriter writer = new BufferedWriter(new FileWriter(outputFile))) {
writer.write("x,y");
writer.newLine();
diff --git a/server/src/main/java/org/apache/iotdb/db/query/eBUG/sample_bottomUpYdiff.java b/server/src/main/java/org/apache/iotdb/db/query/eBUG/sample_bottomUpYdiff.java
index ec96d1d45e8..47c4f1a21e1 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/eBUG/sample_bottomUpYdiff.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/eBUG/sample_bottomUpYdiff.java
@@ -70,7 +70,6 @@ public class sample_bottomUpYdiff {
+ (endTime - startTime)
+ "ms");
- // 输出结果到csv，按照z,x,y三列，因为results结果已经按照z（即DS）递增排序，对应bottom-up的淘汰顺序，越小代表越早被淘汰
try (BufferedWriter writer = new BufferedWriter(new FileWriter(outputFile))) {
writer.write("x,y");
writer.newLine();
diff --git a/server/src/main/java/org/apache/iotdb/db/query/eBUG/sample_eBUG.java b/server/src/main/java/org/apache/iotdb/db/query/eBUG/sample_eBUG.java
index ca44730456c..d8a4b5ea198 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/eBUG/sample_eBUG.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/eBUG/sample_eBUG.java
@@ -65,7 +65,7 @@ public class sample_eBUG {
+ (endTime - startTime)
+ "ms");
- // 输出结果到csv，按照z,x,y三列，因为results结果已经按照z（即DS）递增排序，对应bottom-up的淘汰顺序，越小代表越早被淘汰
+ // 输出结果到csv
try (BufferedWriter writer = new BufferedWriter(new FileWriter(outputFile))) {
// 写入表头
if (m <= 2) {
@@ -75,6 +75,7 @@ public class sample_eBUG {
writer.newLine();
// 写入数据行，按顺序 z, x, y
+ // 按照z,x,y三列，因为results结果已经按照z（即DS）递增排序，对应bottom-up的淘汰顺序，越小代表越早被淘汰
for (Point point : results) {
writer.write(point.z + "," + point.x + "," + point.y);
writer.newLine();