You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by qi...@apache.org on 2020/08/06 04:43:34 UTC
[incubator-iotdb] branch add_load_data_example created (now 8c56016)
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a change to branch add_load_data_example
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git.
at 8c56016 add a data migration example
This branch includes the following new commits:
new 8c56016 add a data migration example
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[incubator-iotdb] 01/01: add a data migration example
Posted by qi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch add_load_data_example
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 8c56016b2df7e5c69480dcd835677e5ab4cb75f8
Author: qiaojialin <64...@qq.com>
AuthorDate: Thu Aug 6 12:42:55 2020 +0800
add a data migration example
---
.../zh/UserGuide/Operation Manual/SQL Reference.md | 2 +-
.../org/apache/iotdb/DataMigrationExample.java | 183 +++++++++++++++++++++
2 files changed, 184 insertions(+), 1 deletion(-)
diff --git a/docs/zh/UserGuide/Operation Manual/SQL Reference.md b/docs/zh/UserGuide/Operation Manual/SQL Reference.md
index ad638a7..93edd9d 100644
--- a/docs/zh/UserGuide/Operation Manual/SQL Reference.md
+++ b/docs/zh/UserGuide/Operation Manual/SQL Reference.md
@@ -766,7 +766,7 @@ Eg: IoTDB > LIST ALL USER OF ROLE roleuser;
ALTER USER <username> SET PASSWORD <password>;
roleName:=identifier
password:=string
-Eg: IoTDB > ALTER USER tempuser SET PASSWORD newpwd;
+Eg: IoTDB > ALTER USER tempuser SET PASSWORD 'newpwd';
```
## 功能
diff --git a/example/session/src/main/java/org/apache/iotdb/DataMigrationExample.java b/example/session/src/main/java/org/apache/iotdb/DataMigrationExample.java
new file mode 100644
index 0000000..d3a22ed
--- /dev/null
+++ b/example/session/src/main/java/org/apache/iotdb/DataMigrationExample.java
@@ -0,0 +1,183 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.SessionDataSet.DataIterator;
+import org.apache.iotdb.session.pool.SessionDataSetWrapper;
+import org.apache.iotdb.session.pool.SessionPool;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.utils.Binary;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+
+
+/**
+ * Migrate all data belongs to a path from one IoTDB to another IoTDB
+ * Each thread migrate one series, the concurrent thread can be configured by concurrency
+ *
+ * This example is
+ * migrating all timeseries from a local IoTDB with 6667 port to a local IoTDB with 6668 port
+ */
+public class DataMigrationExample {
+
+ // used to read data from the source IoTDB
+ private static SessionPool readerPool;
+ // used to write data into the destination IoTDB
+ private static SessionPool writerPool;
+ // concurrent thread of loading timeseries data
+ private static int concurrency = 5;
+
+ public static void main(String[] args)
+ throws IoTDBConnectionException, StatementExecutionException, ExecutionException, InterruptedException {
+
+ ExecutorService executorService = Executors.newFixedThreadPool(2 * concurrency + 1);
+
+ String path = "root";
+
+ if (args.length != 0) {
+ path = args[0];
+ }
+
+ readerPool = new SessionPool("127.0.0.1", 6667, "root", "root", concurrency);
+ writerPool = new SessionPool("127.0.0.1", 6668, "root", "root", concurrency);
+
+ SessionDataSetWrapper schemaDataSet = readerPool.executeQueryStatement("count timeseries " + path);
+ DataIterator schemaIter = schemaDataSet.iterator();
+ int total;
+ if (schemaIter.next()) {
+ total = schemaIter.getInt(1);
+ System.out.println("Total timeseries: " + total);
+ } else {
+ System.out.println("Can not get timeseries schema");
+ System.exit(1);
+ }
+ readerPool.closeResultSet(schemaDataSet);
+
+
+ schemaDataSet = readerPool
+ .executeQueryStatement("show timeseries " + path);
+ schemaIter = schemaDataSet.iterator();
+
+ List<Future> futureList = new ArrayList<>();
+ int count = 0;
+ while (schemaIter.next()) {
+ count ++;
+ Path currentPath = new Path(schemaIter.getString("timeseries"));
+ Future future = executorService.submit(new LoadThread(count, currentPath, TSDataType.valueOf(schemaIter.getString("dataType"))));
+ futureList.add(future);
+ }
+ readerPool.closeResultSet(schemaDataSet);
+
+ for (Future future: futureList) {
+ future.get();
+ }
+ executorService.shutdown();
+
+ readerPool.close();
+ readerPool.close();
+ }
+
+
+ static class LoadThread implements Callable<Void> {
+
+ String device;
+ String measurement;
+ Path series;
+ TSDataType dataType;
+ Tablet tablet;
+ int i;
+
+ public LoadThread(int i, Path series, TSDataType dataType) {
+ this.i = i;
+ this.device = series.getDevice();
+ this.measurement = series.getMeasurement();
+ this.dataType = dataType;
+ this.series = series;
+ List<MeasurementSchema> schemaList = new ArrayList<>();
+ schemaList.add(new MeasurementSchema(measurement, dataType));
+ tablet = new Tablet(device, schemaList, 300000);
+ }
+
+ @Override
+ public Void call() {
+
+ SessionDataSetWrapper dataSet = null;
+
+ try {
+
+ dataSet = readerPool
+ .executeQueryStatement(String.format("select %s from %s", measurement, device));
+
+ DataIterator dataIter = dataSet.iterator();
+ while (dataIter.next()) {
+ int row = tablet.rowSize++;
+ tablet.timestamps[row] = dataIter.getLong(1);
+ switch (dataType) {
+ case BOOLEAN:
+ ((boolean[]) tablet.values[0])[row] = dataIter.getBoolean(2);
+ break;
+ case INT32:
+ ((int[]) tablet.values[0])[row] = dataIter.getInt(2);
+ break;
+ case INT64:
+ ((long[]) tablet.values[0])[row] = dataIter.getLong(2);
+ break;
+ case FLOAT:
+ ((float[]) tablet.values[0])[row] = dataIter.getFloat(2);
+ break;
+ case DOUBLE:
+ ((double[]) tablet.values[0])[row] = dataIter.getDouble(2);
+ break;
+ case TEXT:
+ ((Binary[]) tablet.values[0])[row] = new Binary(dataIter.getString(2));
+ break;
+ }
+ if (tablet.rowSize == tablet.getMaxRowNumber()) {
+ writerPool.insertTablet(tablet, true);
+ tablet.reset();
+ }
+ }
+ if (tablet.rowSize != 0) {
+ writerPool.insertTablet(tablet);
+ tablet.reset();
+ }
+
+ } catch (Exception e) {
+ System.out.println("Loading the " + i + "-th timeseries: " + series + " failed " + e.getMessage());
+ return null;
+ } finally {
+ readerPool.closeResultSet(dataSet);
+ }
+
+ System.out.println("Loading the " + i + "-th timeseries: " + series + " success");
+ return null;
+ }
+
+ }
+}
\ No newline at end of file