You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ha...@apache.org on 2024/05/10 04:30:07 UTC
(iotdb) 02/02: fix implement
This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch import_iot_csv_file_new
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit deba84088e08ac3b7828d7e5e20df573503e4458
Author: HTHou <hh...@outlook.com>
AuthorDate: Fri May 10 12:29:47 2024 +0800
fix implement
---
checkstyle.xml | 7 --
.../main/java/org/apache/iotdb/ImportCSVTool.java | 95 ++++++++++++++++++++--
2 files changed, 87 insertions(+), 15 deletions(-)
diff --git a/checkstyle.xml b/checkstyle.xml
index f1f100e69be..cfcd923d066 100644
--- a/checkstyle.xml
+++ b/checkstyle.xml
@@ -51,13 +51,6 @@
<module name="AvoidStarImport">
<property name="severity" value="error"/>
</module>
- <module name="RegexpSinglelineJava">
- <property name="id" value="unsafeThreadSchedule"/>
- <property name="format" value="schedule(AtFixedRate|WithFixedDelay)\("/>
- <property name="severity" value="error"/>
- <property name="ignoreComments" value="true"/>
- <property name="message" value="Using ScheduledExecutorService::schedule(AtFixedRate|WithFixedDelay) directly is unsafe, please use ScheduledExecutorUtil::(unsafely|safely)Schedule(AtFixedRate|WithFixedDelay) instead."/>
- </module>
<module name="RegexpSinglelineJava">
<property name="id" value="invalidPlanNodeTypeUsage"/>
<property name="format" value="PlanNodeType\.[a-zA-Z_]*\.ordinal\("/>
diff --git a/example/session/src/main/java/org/apache/iotdb/ImportCSVTool.java b/example/session/src/main/java/org/apache/iotdb/ImportCSVTool.java
index 482a23d2819..e59d0bca386 100644
--- a/example/session/src/main/java/org/apache/iotdb/ImportCSVTool.java
+++ b/example/session/src/main/java/org/apache/iotdb/ImportCSVTool.java
@@ -24,11 +24,14 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Stream;
@@ -42,12 +45,18 @@ public class ImportCSVTool {
private static ExecutorService loaderService;
+ private static ScheduledExecutorService logger;
+
private static final String attributesDB = "root.attributes";
private static final Map<String, Long> timeStringCache = new ConcurrentHashMap<>();
private static final Set<String> nameSet = ConcurrentHashMap.newKeySet();
+ private static final ArrayBlockingQueue<Boolean> queue = new ArrayBlockingQueue<>(400);
+
+ private static final AtomicLong totalPoints = new AtomicLong();
+
public static void main(String[] args)
throws IOException, IoTDBConnectionException, StatementExecutionException {
constructRedirectSessionPool(args[0]);
@@ -77,6 +86,20 @@ public class ImportCSVTool {
File folderFile = SystemFileFactory.INSTANCE.getFile(folder);
System.out.println("Start importing!");
long startTime = System.currentTimeMillis();
+
+ logger = Executors.newScheduledThreadPool(1);
+ logger.scheduleAtFixedRate(
+ () -> {
+ long points = totalPoints.get();
+ long time = (System.currentTimeMillis() - startTime) / 1000;
+ double throughout = (double) points / time;
+ System.out.println("Total points: " + points);
+ System.out.println("Total Time: " + points + " ms");
+ System.out.println("Throughout: " + throughout + " points/s");
+ },
+ 0,
+ 1,
+ TimeUnit.MINUTES);
if (folderFile.isDirectory()) {
File[] files =
FSFactoryProducer.getFSFactory().listFilesBySuffix(folderFile.getAbsolutePath(), ".csv");
@@ -102,7 +125,8 @@ public class ImportCSVTool {
"Import "
+ folder
+ " finished. Total cost: "
- + (System.currentTimeMillis() - startTime) + " ms");
+ + (System.currentTimeMillis() - startTime)
+ + " ms");
} else {
if (folderFile.getName().contains("reading")) {
futures.add(
@@ -113,6 +137,7 @@ public class ImportCSVTool {
loaderService.submit(
() -> importCsvDiagnostics(folderFile, folderFile.getName().contains("null"))));
}
+
for (Future<?> task : futures) {
try {
task.get();
@@ -124,10 +149,12 @@ public class ImportCSVTool {
"Import "
+ folder
+ " finished. Total cost "
- + (System.currentTimeMillis() - startTime) + " ms");
+ + (System.currentTimeMillis() - startTime)
+ + " ms");
}
loaderService.shutdown();
sessionService.shutdown();
+ logger.shutdown();
}
private static void constructRedirectSessionPool(String host) {
@@ -203,14 +230,26 @@ public class ImportCSVTool {
}
if (currentDeviceId[0] != null) {
Tablet oldTablet = tabletMap.remove(currentDeviceId[0]);
+ try {
+ queue.put(true);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
sessionService.submit(
() -> {
try {
sessionPool.insertAlignedTablet(oldTablet);
+ totalPoints.addAndGet(
+ (long) oldTablet.rowSize * oldTablet.getSchemas().size());
} catch (IoTDBConnectionException | StatementExecutionException e) {
throw new RuntimeException(e);
+ } finally {
+ try {
+ queue.take();
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
}
- oldTablet.reset();
});
}
currentDeviceId[0] = deviceId;
@@ -218,14 +257,26 @@ public class ImportCSVTool {
Tablet tablet = tabletMap.get(currentDeviceId[0]);
if (tablet.rowSize == tablet.getMaxRowNumber()) {
Tablet oldTablet = tabletMap.remove(currentDeviceId[0]);
+ try {
+ queue.put(true);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
sessionService.submit(
() -> {
try {
sessionPool.insertAlignedTablet(oldTablet);
+ totalPoints.addAndGet(
+ (long) oldTablet.rowSize * oldTablet.getSchemas().size());
} catch (IoTDBConnectionException | StatementExecutionException e) {
throw new RuntimeException(e);
+ } finally {
+ try {
+ queue.take();
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
}
- oldTablet.reset();
});
}
}
@@ -247,13 +298,15 @@ public class ImportCSVTool {
+ " Progress: "
+ currentProgress.get()
+ "% cost "
- + (System.currentTimeMillis() - startTime) + " ms");
+ + (System.currentTimeMillis() - startTime)
+ + " ms");
}
}
});
if (!tabletMap.isEmpty()) {
for (Tablet tablet : tabletMap.values()) {
sessionPool.insertAlignedTablet(tablet);
+ totalPoints.addAndGet((long) tablet.rowSize * tablet.getSchemas().size());
}
tabletMap.clear();
}
@@ -320,14 +373,26 @@ public class ImportCSVTool {
}
if (currentDeviceId[0] != null) {
Tablet oldTablet = tabletMap.remove(currentDeviceId[0]);
+ try {
+ queue.put(true);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
sessionService.submit(
() -> {
try {
sessionPool.insertAlignedTablet(oldTablet);
+ totalPoints.addAndGet(
+ (long) oldTablet.rowSize * oldTablet.getSchemas().size());
} catch (IoTDBConnectionException | StatementExecutionException e) {
throw new RuntimeException(e);
+ } finally {
+ try {
+ queue.take();
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
}
- oldTablet.reset();
});
}
currentDeviceId[0] = deviceId;
@@ -335,14 +400,26 @@ public class ImportCSVTool {
Tablet tablet = tabletMap.get(currentDeviceId[0]);
if (tablet.rowSize == tablet.getMaxRowNumber()) {
Tablet oldTablet = tabletMap.remove(currentDeviceId[0]);
+ try {
+ queue.put(true);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
sessionService.submit(
() -> {
try {
sessionPool.insertAlignedTablet(oldTablet);
+ totalPoints.addAndGet(
+ (long) oldTablet.rowSize * oldTablet.getSchemas().size());
} catch (IoTDBConnectionException | StatementExecutionException e) {
throw new RuntimeException(e);
+ } finally {
+ try {
+ queue.take();
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
}
- oldTablet.reset();
});
}
}
@@ -364,13 +441,15 @@ public class ImportCSVTool {
+ " Progress: "
+ currentProgress.get()
+ "% cost "
- + (System.currentTimeMillis() - startTime) + " ms");
+ + (System.currentTimeMillis() - startTime)
+ + " ms");
}
}
});
if (!tabletMap.isEmpty()) {
for (Tablet tablet : tabletMap.values()) {
sessionPool.insertAlignedTablet(tablet);
+ totalPoints.addAndGet((long) tablet.rowSize * tablet.getSchemas().size());
}
tabletMap.clear();
}