This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 1578144 use jdbc to connect iotdb in spark (#381)
1578144 is described below
commit 1578144cdb5bbc029dc69c41327820bc37215395
Author: SilverNarcissus <[email protected]>
AuthorDate: Thu Sep 12 17:06:54 2019 +0800
use jdbc to connect iotdb in spark (#381)
* use jdbc to connect iotdb in spark
---
.../{9-Tools-spark.md => 9-Tools-spark-iotdb.md} | 0
.../{9-Tools-spark.md => 9-Tools-spark-tsfile.md} | 0
.../Documentation/UserGuide/9-Tools-spark-iotdb.md | 171 +++++++++++++
.../{9-Tools-spark.md => 9-Tools-spark-tsfile.md} | 0
pom.xml | 1 +
spark-iotdb-connector/Readme.md | 171 +++++++++++++
spark-iotdb-connector/pom.xml | 135 ++++++++++
.../java/org/apache/iotdb/sparkdb/SQLConstant.java | 31 +++
.../scala/org/apache/iotdb/sparkdb/Converter.scala | 87 +++++++
.../org/apache/iotdb/sparkdb/DefaultSource.scala | 42 +++
.../org/apache/iotdb/sparkdb/IoTDBOptions.scala | 42 +++
.../scala/org/apache/iotdb/sparkdb/IoTDBRDD.scala | 134 ++++++++++
.../org/apache/iotdb/sparkdb/IoTDBRelation.scala | 116 +++++++++
.../org/apache/iotdb/sparkdb/Transformer.scala | 173 +++++++++++++
.../scala/org/apache/iotdb/sparkdb/package.scala | 35 +++
.../org/apache/iotdb/sparkdb/EnvironmentUtils.java | 282 +++++++++++++++++++++
.../scala/org/apache/iotdb/sparkdb/IoTDBTest.scala | 115 +++++++++
17 files changed, 1535 insertions(+)
diff --git a/docs/Documentation-CHN/UserGuide/9-Tools-spark.md
b/docs/Documentation-CHN/UserGuide/9-Tools-spark-iotdb.md
similarity index 100%
copy from docs/Documentation-CHN/UserGuide/9-Tools-spark.md
copy to docs/Documentation-CHN/UserGuide/9-Tools-spark-iotdb.md
diff --git a/docs/Documentation-CHN/UserGuide/9-Tools-spark.md
b/docs/Documentation-CHN/UserGuide/9-Tools-spark-tsfile.md
similarity index 100%
rename from docs/Documentation-CHN/UserGuide/9-Tools-spark.md
rename to docs/Documentation-CHN/UserGuide/9-Tools-spark-tsfile.md
diff --git a/docs/Documentation/UserGuide/9-Tools-spark-iotdb.md
b/docs/Documentation/UserGuide/9-Tools-spark-iotdb.md
new file mode 100644
index 0000000..581baac
--- /dev/null
+++ b/docs/Documentation/UserGuide/9-Tools-spark-iotdb.md
@@ -0,0 +1,171 @@
+<!--
+
+ 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.
+
+-->
+## version
+
+The versions required for Spark and Java are as follow:
+
+| Spark Version | Scala Version | Java Version | TsFile |
+| ------------- | ------------- | ------------ |------------ |
+| `2.4.3` | `2.11` | `1.8` | `0.9.0-SNAPSHOT`|
+
+
+## install
+mvn clean scala:compile compile install
+
+
+# 1. maven dependency
+
+```
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>spark-iotdb-connector</artifactId>
+ <version>0.9.0-SNAPSHOT</version>
+ </dependency>
+```
+
+
+# 2. spark-shell user guide
+
+```
+spark-shell --jars
spark-iotdb-connector-0.9.0-SNAPSHOT.jar,iotdb-jdbc-0.9.0-SNAPSHOT-jar-with-dependencies.jar
+
+import org.apache.iotdb.sparkdb._
+
+val df =
spark.read.format("org.apache.iotdb.sparkdb").option("url","jdbc:iotdb://127.0.0.1:6667/").option("sql","select
* from root").load
+
+df.printSchema()
+
+df.show()
+```
+
+### if you want to partition your rdd, you can do as following
+```
+spark-shell --jars
spark-iotdb-connector-0.9.0-SNAPSHOT.jar,iotdb-jdbc-0.9.0-SNAPSHOT-jar-with-dependencies.jar
+
+import org.apache.iotdb.sparkdb._
+
+val df =
spark.read.format("org.apache.iotdb.sparkdb").option("url","jdbc:iotdb://127.0.0.1:6667/").option("sql","select
* from root").
+ option("lowerBound", [lower bound of time that you
want query(include)]).option("upperBound", [upper bound of time that you want
query(include)]).
+ option("numPartition", [the partition number you
want]).load
+
+df.printSchema()
+
+df.show()
+```
+
+# 3. Schema Inference
+
+Take the following TsFile structure as an example: There are three
Measurements in the TsFile schema: status, temperature, and hardware. The basic
information of these three measurements is as follows:
+
+<center>
+<table style="text-align:center">
+ <tr><th colspan="2">Name</th><th colspan="2">Type</th><th
colspan="2">Encode</th></tr>
+ <tr><td colspan="2">status</td><td colspan="2">Boolean</td><td
colspan="2">PLAIN</td></tr>
+ <tr><td colspan="2">temperature</td><td colspan="2">Float</td><td
colspan="2">RLE</td></tr>
+ <tr><td colspan="2">hardware</td><td colspan="2">Text</td><td
colspan="2">PLAIN</td></tr>
+</table>
+</center>
+
+The existing data in the TsFile is as follows:
+
+
+<center>
+<table style="text-align:center">
+ <tr><th colspan="4">device:root.ln.wf01.wt01</th><th
colspan="4">device:root.ln.wf02.wt02</th></tr>
+ <tr><th colspan="2">status</th><th colspan="2">temperature</th><th
colspan="2">hardware</th><th colspan="2">status</th></tr>
+
<tr><th>time</th><th>value</td><th>time</th><th>value</td><th>time</th><th>value</th><th>time</th><th>value</td></tr>
+
<tr><td>1</td><td>True</td><td>1</td><td>2.2</td><td>2</td><td>"aaa"</td><td>1</td><td>True</td></tr>
+
<tr><td>3</td><td>True</td><td>2</td><td>2.2</td><td>4</td><td>"bbb"</td><td>2</td><td>False</td></tr>
+ <tr><td>5</td><td> False
</td><td>3</td><td>2.1</td><td>6</td><td>"ccc"</td><td>4</td><td>True</td></tr>
+</table>
+</center>
+
+
+The wide(default) table form is as follows:
+
+| time | root.ln.wf02.wt02.temperature | root.ln.wf02.wt02.status |
root.ln.wf02.wt02.hardware | root.ln.wf01.wt01.temperature |
root.ln.wf01.wt01.status | root.ln.wf01.wt01.hardware |
+|------|-------------------------------|--------------------------|----------------------------|-------------------------------|--------------------------|----------------------------|
+| 1 | null | true | null
| 2.2 | true |
null |
+| 2 | null | false | aaa
| 2.2 | null |
null |
+| 3 | null | null | null
| 2.1 | true |
null |
+| 4 | null | true | bbb
| null | null |
null |
+| 5 | null | null | null
| null | false |
null |
+| 6 | null | null | ccc
| null | null |
null |
+
+You can also use narrow table form which as follows: (You can see part 4 about
how to use narrow form)
+
+| time | device_name | status | hardware
| temperature |
+|------|-------------------------------|--------------------------|----------------------------|-------------------------------|
+| 1 | root.ln.wf02.wt01 | true | null
| 2.2 |
+| 1 | root.ln.wf02.wt02 | true | null
| null |
+| 2 | root.ln.wf02.wt01 | null | null
| 2.2 |
+| 2 | root.ln.wf02.wt02 | false | aaa
| null |
+| 3 | root.ln.wf02.wt01 | true | null
| 2.1 |
+| 4 | root.ln.wf02.wt02 | true | bbb
| null |
+| 5 | root.ln.wf02.wt01 | false | null
| null |
+| 6 | root.ln.wf02.wt02 | null | ccc
| null |
+
+# 4. Transform between wide and narrow table
+
+## from wide to narrow
+```
+import org.apache.iotdb.sparkdb._
+
+val wide_df = spark.read.format("org.apache.iotdb.sparkdb").option("url",
"jdbc:iotdb://127.0.0.1:6667/").option("sql", "select * from root where time <
1100 and time > 1000").load
+val narrow_df = Transformer.toNarrowForm(spark, wide_df)
+```
+
+## from narrow to wide
+```
+import org.apache.iotdb.sparkdb._
+
+val wide_df = Transformer.toWideForm(spark, narrow_df)
+```
+
+# 5. Java user guide
+```
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+import org.apache.iotdb.sparkdb.*
+
+public class Example {
+
+ public static void main(String[] args) {
+ SparkSession spark = SparkSession
+ .builder()
+ .appName("Build a DataFrame from Scratch")
+ .master("local[*]")
+ .getOrCreate();
+
+ Dataset<Row> df = spark.read().format("org.apache.iotdb.sparkdb")
+ .option("url","jdbc:iotdb://127.0.0.1:6667/")
+ .option("sql","select * from root").load();
+
+ df.printSchema();
+
+ df.show();
+
+ Dataset<Row> narrowTable = Transformer.toNarrowForm(spark, df)
+ narrowTable.show()
+ }
+}
+```
\ No newline at end of file
diff --git a/docs/Documentation/UserGuide/9-Tools-spark.md
b/docs/Documentation/UserGuide/9-Tools-spark-tsfile.md
similarity index 100%
rename from docs/Documentation/UserGuide/9-Tools-spark.md
rename to docs/Documentation/UserGuide/9-Tools-spark-tsfile.md
diff --git a/pom.xml b/pom.xml
index 77a2219..38f69f6 100644
--- a/pom.xml
+++ b/pom.xml
@@ -54,6 +54,7 @@
<module>example</module>
<module>grafana</module>
<module>spark-tsfile</module>
+ <module>spark-iotdb-connector</module>
<!-- <module>hadoop</module> -->
<module>distribution</module>
</modules>
diff --git a/spark-iotdb-connector/Readme.md b/spark-iotdb-connector/Readme.md
new file mode 100644
index 0000000..581baac
--- /dev/null
+++ b/spark-iotdb-connector/Readme.md
@@ -0,0 +1,171 @@
+<!--
+
+ 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.
+
+-->
+## version
+
+The versions required for Spark and Java are as follow:
+
+| Spark Version | Scala Version | Java Version | TsFile |
+| ------------- | ------------- | ------------ |------------ |
+| `2.4.3` | `2.11` | `1.8` | `0.9.0-SNAPSHOT`|
+
+
+## install
+mvn clean scala:compile compile install
+
+
+# 1. maven dependency
+
+```
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>spark-iotdb-connector</artifactId>
+ <version>0.9.0-SNAPSHOT</version>
+ </dependency>
+```
+
+
+# 2. spark-shell user guide
+
+```
+spark-shell --jars
spark-iotdb-connector-0.9.0-SNAPSHOT.jar,iotdb-jdbc-0.9.0-SNAPSHOT-jar-with-dependencies.jar
+
+import org.apache.iotdb.sparkdb._
+
+val df =
spark.read.format("org.apache.iotdb.sparkdb").option("url","jdbc:iotdb://127.0.0.1:6667/").option("sql","select
* from root").load
+
+df.printSchema()
+
+df.show()
+```
+
+### if you want to partition your rdd, you can do as following
+```
+spark-shell --jars
spark-iotdb-connector-0.9.0-SNAPSHOT.jar,iotdb-jdbc-0.9.0-SNAPSHOT-jar-with-dependencies.jar
+
+import org.apache.iotdb.sparkdb._
+
+val df =
spark.read.format("org.apache.iotdb.sparkdb").option("url","jdbc:iotdb://127.0.0.1:6667/").option("sql","select
* from root").
+ option("lowerBound", [lower bound of time that you
want query(include)]).option("upperBound", [upper bound of time that you want
query(include)]).
+ option("numPartition", [the partition number you
want]).load
+
+df.printSchema()
+
+df.show()
+```
+
+# 3. Schema Inference
+
+Take the following TsFile structure as an example: There are three
Measurements in the TsFile schema: status, temperature, and hardware. The basic
information of these three measurements is as follows:
+
+<center>
+<table style="text-align:center">
+ <tr><th colspan="2">Name</th><th colspan="2">Type</th><th
colspan="2">Encode</th></tr>
+ <tr><td colspan="2">status</td><td colspan="2">Boolean</td><td
colspan="2">PLAIN</td></tr>
+ <tr><td colspan="2">temperature</td><td colspan="2">Float</td><td
colspan="2">RLE</td></tr>
+ <tr><td colspan="2">hardware</td><td colspan="2">Text</td><td
colspan="2">PLAIN</td></tr>
+</table>
+</center>
+
+The existing data in the TsFile is as follows:
+
+
+<center>
+<table style="text-align:center">
+ <tr><th colspan="4">device:root.ln.wf01.wt01</th><th
colspan="4">device:root.ln.wf02.wt02</th></tr>
+ <tr><th colspan="2">status</th><th colspan="2">temperature</th><th
colspan="2">hardware</th><th colspan="2">status</th></tr>
+
<tr><th>time</th><th>value</td><th>time</th><th>value</td><th>time</th><th>value</th><th>time</th><th>value</td></tr>
+
<tr><td>1</td><td>True</td><td>1</td><td>2.2</td><td>2</td><td>"aaa"</td><td>1</td><td>True</td></tr>
+
<tr><td>3</td><td>True</td><td>2</td><td>2.2</td><td>4</td><td>"bbb"</td><td>2</td><td>False</td></tr>
+ <tr><td>5</td><td> False
</td><td>3</td><td>2.1</td><td>6</td><td>"ccc"</td><td>4</td><td>True</td></tr>
+</table>
+</center>
+
+
+The wide(default) table form is as follows:
+
+| time | root.ln.wf02.wt02.temperature | root.ln.wf02.wt02.status |
root.ln.wf02.wt02.hardware | root.ln.wf01.wt01.temperature |
root.ln.wf01.wt01.status | root.ln.wf01.wt01.hardware |
+|------|-------------------------------|--------------------------|----------------------------|-------------------------------|--------------------------|----------------------------|
+| 1 | null | true | null
| 2.2 | true |
null |
+| 2 | null | false | aaa
| 2.2 | null |
null |
+| 3 | null | null | null
| 2.1 | true |
null |
+| 4 | null | true | bbb
| null | null |
null |
+| 5 | null | null | null
| null | false |
null |
+| 6 | null | null | ccc
| null | null |
null |
+
+You can also use narrow table form which as follows: (You can see part 4 about
how to use narrow form)
+
+| time | device_name | status | hardware
| temperature |
+|------|-------------------------------|--------------------------|----------------------------|-------------------------------|
+| 1 | root.ln.wf02.wt01 | true | null
| 2.2 |
+| 1 | root.ln.wf02.wt02 | true | null
| null |
+| 2 | root.ln.wf02.wt01 | null | null
| 2.2 |
+| 2 | root.ln.wf02.wt02 | false | aaa
| null |
+| 3 | root.ln.wf02.wt01 | true | null
| 2.1 |
+| 4 | root.ln.wf02.wt02 | true | bbb
| null |
+| 5 | root.ln.wf02.wt01 | false | null
| null |
+| 6 | root.ln.wf02.wt02 | null | ccc
| null |
+
+# 4. Transform between wide and narrow table
+
+## from wide to narrow
+```
+import org.apache.iotdb.sparkdb._
+
+val wide_df = spark.read.format("org.apache.iotdb.sparkdb").option("url",
"jdbc:iotdb://127.0.0.1:6667/").option("sql", "select * from root where time <
1100 and time > 1000").load
+val narrow_df = Transformer.toNarrowForm(spark, wide_df)
+```
+
+## from narrow to wide
+```
+import org.apache.iotdb.sparkdb._
+
+val wide_df = Transformer.toWideForm(spark, narrow_df)
+```
+
+# 5. Java user guide
+```
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+import org.apache.iotdb.sparkdb.*
+
+public class Example {
+
+ public static void main(String[] args) {
+ SparkSession spark = SparkSession
+ .builder()
+ .appName("Build a DataFrame from Scratch")
+ .master("local[*]")
+ .getOrCreate();
+
+ Dataset<Row> df = spark.read().format("org.apache.iotdb.sparkdb")
+ .option("url","jdbc:iotdb://127.0.0.1:6667/")
+ .option("sql","select * from root").load();
+
+ df.printSchema();
+
+ df.show();
+
+ Dataset<Row> narrowTable = Transformer.toNarrowForm(spark, df)
+ narrowTable.show()
+ }
+}
+```
\ No newline at end of file
diff --git a/spark-iotdb-connector/pom.xml b/spark-iotdb-connector/pom.xml
new file mode 100644
index 0000000..a18120d
--- /dev/null
+++ b/spark-iotdb-connector/pom.xml
@@ -0,0 +1,135 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-parent</artifactId>
+ <version>0.9.0-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+ <artifactId>spark-iotdb-connector</artifactId>
+ <version>0.9.0-SNAPSHOT</version>
+ <packaging>jar</packaging>
+ <properties>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ <compile.version>1.8</compile.version>
+ </properties>
+ <dependencies>
+ <dependency>
+ <groupId>org.json</groupId>
+ <artifactId>json</artifactId>
+ <version>20170516</version>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <version>${junit.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-jdbc</artifactId>
+ <version>0.9.0-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-server</artifactId>
+ <version>0.9.0-SNAPSHOT</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-core_2.11</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-sql_2.11</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.scala-lang</groupId>
+ <artifactId>scala-library</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.scalatest</groupId>
+ <artifactId>scalatest_2.11</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>net.alchim31.maven</groupId>
+ <artifactId>scala-maven-plugin</artifactId>
+ <version>3.2.0</version>
+ <executions>
+ <execution>
+ <id>compile-scala</id>
+ <phase>compile</phase>
+ <goals>
+ <goal>add-source</goal>
+ <goal>compile</goal>
+ </goals>
+ </execution>
+ <execution>
+ <id>test-compile-scala</id>
+ <phase>test-compile</phase>
+ <goals>
+ <goal>add-source</goal>
+ <goal>testCompile</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ <profiles>
+ <profile>
+ <id>java-11-and-above</id>
+ <activation>
+ <!-- This needs to be updated as soon as Java 20 is shipped -->
+ <jdk>[11,20)</jdk>
+ </activation>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>net.alchim31.maven</groupId>
+ <artifactId>scala-maven-plugin</artifactId>
+ <version>3.2.0</version>
+ <!-- Added to avoid problems with Java 11 and above -->
+ <configuration>
+ <args>
+ <arg>-nobootcp</arg>
+ </args>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
+</project>
diff --git
a/spark-iotdb-connector/src/main/java/org/apache/iotdb/sparkdb/SQLConstant.java
b/spark-iotdb-connector/src/main/java/org/apache/iotdb/sparkdb/SQLConstant.java
new file mode 100644
index 0000000..466df21
--- /dev/null
+++
b/spark-iotdb-connector/src/main/java/org/apache/iotdb/sparkdb/SQLConstant.java
@@ -0,0 +1,31 @@
+/**
+ * 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.sparkdb;
+
+/**
+ * this class contains several constants used in SQL.
+ */
+public class SQLConstant {
+
+ public static final String NEED_NOT_TO_PRINT_TIMESTAMP = "AGGREGATION";
+ public static final String RESERVED_TIME = "time";
+ public static final String TIMESTAMP_STR = "Time";
+ public static final String NULL_STR = "null";
+ public static final String WHERE = "where";
+}
diff --git
a/spark-iotdb-connector/src/main/scala/org/apache/iotdb/sparkdb/Converter.scala
b/spark-iotdb-connector/src/main/scala/org/apache/iotdb/sparkdb/Converter.scala
new file mode 100644
index 0000000..9548976
--- /dev/null
+++
b/spark-iotdb-connector/src/main/scala/org/apache/iotdb/sparkdb/Converter.scala
@@ -0,0 +1,87 @@
+/**
+ * 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.sparkdb
+
+import java.sql._
+
+import org.apache.spark.sql.types._
+import org.slf4j.LoggerFactory
+import java.sql.Statement
+
+import scala.collection.mutable.ListBuffer
+
+class Converter
+
+object Converter {
+ private final val logger = LoggerFactory.getLogger(classOf[Converter])
+
+ def toSqlData(field: StructField, value: String): Any = {
+ if (value == null || value.equals(SQLConstant.NULL_STR)) return null
+
+ val r = field.dataType match {
+ case BooleanType => java.lang.Boolean.valueOf(value)
+ case IntegerType => value.toInt
+ case LongType => value.toLong
+ case FloatType => value.toFloat
+ case DoubleType => value.toDouble
+ case StringType => value
+ case other => throw new UnsupportedOperationException(s"Unsupported type
$other")
+ }
+ r
+ }
+
+ def toSparkSchema(options: IoTDBOptions): StructType = {
+
+ Class.forName("org.apache.iotdb.jdbc.IoTDBDriver")
+ val sqlConn: Connection = DriverManager.getConnection(options.url,
options.user, options.password)
+ val sqlStatement: Statement = sqlConn.createStatement()
+ val hasResultSet: Boolean = sqlStatement.execute(options.sql)
+
+ val fields = new ListBuffer[StructField]()
+ if (hasResultSet) {
+ val resultSet: ResultSet = sqlStatement.getResultSet
+ val resultSetMetaData: ResultSetMetaData = resultSet.getMetaData
+
+ if (resultSetMetaData.getColumnTypeName(0) != null) {
+ val printTimestamp: Boolean =
!resultSetMetaData.getColumnTypeName(0).toUpperCase().equals(SQLConstant.NEED_NOT_TO_PRINT_TIMESTAMP)
+ if (printTimestamp) {
+ fields += StructField(SQLConstant.TIMESTAMP_STR, LongType, nullable
= false)
+ }
+ }
+
+ val colCount = resultSetMetaData.getColumnCount
+
+ for (i <- 2 to colCount) {
+ fields += StructField(resultSetMetaData.getColumnLabel(i),
resultSetMetaData.getColumnType(i) match {
+ case Types.BOOLEAN => BooleanType
+ case Types.INTEGER => IntegerType
+ case Types.BIGINT => LongType
+ case Types.FLOAT => FloatType
+ case Types.DOUBLE => DoubleType
+ case Types.VARCHAR => StringType
+ case other => throw new UnsupportedOperationException(s"Unsupported
type $other")
+ }, nullable = true)
+ }
+ StructType(fields.toList)
+ }
+ else {
+ StructType(fields)
+ }
+ }
+}
diff --git
a/spark-iotdb-connector/src/main/scala/org/apache/iotdb/sparkdb/DefaultSource.scala
b/spark-iotdb-connector/src/main/scala/org/apache/iotdb/sparkdb/DefaultSource.scala
new file mode 100644
index 0000000..cf82c7d
--- /dev/null
+++
b/spark-iotdb-connector/src/main/scala/org/apache/iotdb/sparkdb/DefaultSource.scala
@@ -0,0 +1,42 @@
+/**
+ * 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.sparkdb
+
+import org.apache.spark.sql.SQLContext
+import org.apache.spark.sql.sources.{BaseRelation, DataSourceRegister,
RelationProvider}
+import org.slf4j.LoggerFactory
+
+private[iotdb] class DefaultSource extends RelationProvider with
DataSourceRegister {
+ private final val logger = LoggerFactory.getLogger(classOf[DefaultSource])
+
+ override def shortName(): String = "tsfile"
+
+ override def createRelation(
+ sqlContext: SQLContext,
+ parameters: Map[String, String]): BaseRelation
= {
+
+ val iotdbOptions = new IoTDBOptions(parameters)
+
+ if (iotdbOptions.url == null || iotdbOptions.sql == null) {
+ sys.error("IoTDB url or sql not specified")
+ }
+ new IoTDBRelation(iotdbOptions)(sqlContext.sparkSession)
+
+ }
+}
\ No newline at end of file
diff --git
a/spark-iotdb-connector/src/main/scala/org/apache/iotdb/sparkdb/IoTDBOptions.scala
b/spark-iotdb-connector/src/main/scala/org/apache/iotdb/sparkdb/IoTDBOptions.scala
new file mode 100644
index 0000000..c595ed5
--- /dev/null
+++
b/spark-iotdb-connector/src/main/scala/org/apache/iotdb/sparkdb/IoTDBOptions.scala
@@ -0,0 +1,42 @@
+/**
+ * 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.sparkdb
+
+class IoTDBOptions(
+ @transient private val parameters: Map[String, String])
+ extends Serializable {
+
+ val url = parameters.getOrElse("url", sys.error("Option 'url' not
specified"))
+
+ val user = parameters.getOrElse("user", "root")
+
+ val password = parameters.getOrElse("password", "root")
+
+ val sql = parameters.getOrElse("sql", sys.error("Option 'sql' not
specified"))
+
+ val numPartition = parameters.getOrElse("numPartition", "1")
+
+ val lowerBound = parameters.getOrElse("lowerBound", "0")
+
+ val upperBound = parameters.getOrElse("upperBound", "0")
+
+ def get(name: String): Unit = {
+
+ }
+}
diff --git
a/spark-iotdb-connector/src/main/scala/org/apache/iotdb/sparkdb/IoTDBRDD.scala
b/spark-iotdb-connector/src/main/scala/org/apache/iotdb/sparkdb/IoTDBRDD.scala
new file mode 100644
index 0000000..1c45dc8
--- /dev/null
+++
b/spark-iotdb-connector/src/main/scala/org/apache/iotdb/sparkdb/IoTDBRDD.scala
@@ -0,0 +1,134 @@
+/**
+ * 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.sparkdb
+
+import org.apache.spark.{Partition, SparkContext, TaskContext}
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.sources._
+import java.sql.{Connection, DriverManager, ResultSet, Statement}
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.types._
+
+
+//IoTDB data partition
+case class IoTDBPartition(where: String, id: Int, start: java.lang.Long, end:
java.lang.Long) extends Partition {
+ override def index: Int = id
+}
+
+object IoTDBRDD {
+
+ private def pruneSchema(schema: StructType, columns: Array[String]):
StructType = {
+ val fieldMap = Map(schema.fields.map(x => x.name -> x): _*)
+ new StructType(columns.map(name => fieldMap(name)))
+ }
+
+}
+
+class IoTDBRDD private[iotdb](
+ sc: SparkContext,
+ options: IoTDBOptions,
+ schema: StructType,
+ requiredColumns: Array[String],
+ filters: Array[Filter],
+ partitions: Array[Partition])
+ extends RDD[Row](sc, Nil) {
+
+ override def compute(split: Partition, context: TaskContext): Iterator[Row]
= new Iterator[Row] {
+ var finished = false
+ var gotNext = false
+ var nextValue: Row = _
+ val inputMetrics = context.taskMetrics().inputMetrics
+
+ val part = split.asInstanceOf[IoTDBPartition]
+
+ var taskInfo: String = _
+ Option(TaskContext.get()).foreach { taskContext => {
+ taskContext.addTaskCompletionListener { _ => conn.close() }
+ taskInfo = "task Id: " + taskContext.taskAttemptId() + " partition Id: "
+ taskContext.partitionId()
+ }
+ }
+
+ Class.forName("org.apache.iotdb.jdbc.IoTDBDriver")
+ val conn: Connection = DriverManager.getConnection(options.url,
options.user, options.password)
+ val stmt: Statement = conn.createStatement()
+
+ var sql = options.sql
+ // for different partition
+ if (part.where != null) {
+ val sqlPart = options.sql.split(SQLConstant.WHERE)
+ sql = sqlPart(0) + " " + SQLConstant.WHERE + " (" + part.where + ") "
+ if (sqlPart.length == 2) {
+ sql += "and (" + sqlPart(1) + ")"
+ }
+ }
+ //
+ var rs: ResultSet = stmt.executeQuery(sql)
+ val prunedSchema = IoTDBRDD.pruneSchema(schema, requiredColumns)
+ private val rowBuffer = Array.fill[Any](prunedSchema.length)(null)
+
+ def getNext: Row = {
+ if (rs.next()) {
+ val fields = new scala.collection.mutable.HashMap[String, String]()
+ for (i <- 1 until rs.getMetaData.getColumnCount + 1) {
+ // start from 1
+ val field = rs.getString(i)
+ fields.put(rs.getMetaData.getColumnName(i), field)
+ }
+
+ //index in one required row
+ var index = 0
+ prunedSchema.foreach((field: StructField) => {
+ val r = Converter.toSqlData(field, fields.getOrElse(field.name,
null))
+ rowBuffer(index) = r
+ index += 1
+ })
+ Row.fromSeq(rowBuffer)
+
+ }
+ else {
+ finished = true
+ null
+ }
+ }
+
+
+ override def hasNext: Boolean = {
+ if (!finished) {
+ if (!gotNext) {
+ nextValue = getNext
+ gotNext = true
+ }
+ }
+ !finished
+ }
+
+ override def next(): Row = {
+ if (!hasNext) {
+ throw new NoSuchElementException("End of stream")
+ }
+ gotNext = false
+ nextValue
+ }
+ }
+
+ override def getPartitions: Array[Partition] = partitions
+
+
+}
\ No newline at end of file
diff --git
a/spark-iotdb-connector/src/main/scala/org/apache/iotdb/sparkdb/IoTDBRelation.scala
b/spark-iotdb-connector/src/main/scala/org/apache/iotdb/sparkdb/IoTDBRelation.scala
new file mode 100644
index 0000000..bf30ebb
--- /dev/null
+++
b/spark-iotdb-connector/src/main/scala/org/apache/iotdb/sparkdb/IoTDBRelation.scala
@@ -0,0 +1,116 @@
+/**
+ * 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.sparkdb
+
+import org.apache.spark.Partition
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.sources.{BaseRelation, Filter, PrunedFilteredScan}
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.{Row, SQLContext, SparkSession}
+import org.slf4j.LoggerFactory
+
+import scala.collection.mutable.ArrayBuffer
+
+private case class IoTDBPartitioningInfo(
+ start: Long,
+ end: Long,
+ numPartitions: Int)
+
+private object IoTDBRelation {
+
+ private final val logger = LoggerFactory.getLogger(classOf[IoTDBRelation])
+
+ def getPartitions(partitionInfo: IoTDBPartitioningInfo): Array[Partition] = {
+ if (partitionInfo == null || partitionInfo.numPartitions <= 1 ||
+ partitionInfo.start == partitionInfo.end) {
+ return Array[Partition](IoTDBPartition(null, 0, 0L, 0L))
+ }
+ val start = partitionInfo.start
+ val end = partitionInfo.end
+
+ //if start <= end , can not partition
+ require(start <= end,
+ "Operation not allowed: the start time is larger than end time " +
+ s"time start: $start; end: $end")
+
+ //numPartitions needs to be less and equal than (end - start)
+ val numPartitions =
+ if ((end - start) >= partitionInfo.numPartitions) {
+ partitionInfo.numPartitions
+ } else {
+ logger.warn("The number of partitions is reduced because the specified
number of " +
+ "partitions is less than the difference between upper bound and lower
bound. " +
+ s"Updated number of partitions: ${end - start}; Input number of " +
+ s"partitions: ${partitionInfo.numPartitions}; Lower bound: $start; " +
+ s"Upper bound: $end.")
+ end - start
+ }
+
+ var partitions = new ArrayBuffer[Partition]()
+
+ val length: Long = (end - start) / numPartitions + 1
+ var i: Int = 0
+ var currentValue: Long = start
+ while (i < numPartitions) {
+ var where = s""
+ if (i == 0) {
+
+ where = s"${SQLConstant.RESERVED_TIME} >= $currentValue and
${SQLConstant.RESERVED_TIME} <= ${currentValue + length}"
+ partitions += IoTDBPartition(where, i, currentValue, currentValue +
length)
+ }
+ else {
+ where = s"${SQLConstant.RESERVED_TIME} > $currentValue and
${SQLConstant.RESERVED_TIME} <= ${currentValue + length}"
+ partitions += IoTDBPartition(where, i, currentValue + 1, currentValue
+ length)
+ }
+
+ i = i + 1
+ currentValue += length
+ }
+ partitions.toArray
+ }
+}
+
+class IoTDBRelation protected[iotdb](val options: IoTDBOptions)(@transient val
sparkSession: SparkSession)
+ extends BaseRelation with PrunedFilteredScan {
+
+ override def sqlContext: SQLContext = sparkSession.sqlContext
+
+ private final val logger = LoggerFactory.getLogger(classOf[IoTDBRelation])
+
+ override def schema: StructType = {
+ Converter.toSparkSchema(options)
+ }
+
+ override def buildScan(requiredColumns: Array[String], filters:
Array[Filter]): RDD[Row] = {
+ val start: Long = options.lowerBound.toLong
+ val end: Long = options.upperBound.toLong
+ val numPartition = options.numPartition.toInt
+
+ val partitionInfo = IoTDBPartitioningInfo(start, end, numPartition)
+
+ val parts = IoTDBRelation.getPartitions(partitionInfo)
+
+ new IoTDBRDD(sparkSession.sparkContext,
+ options,
+ schema,
+ requiredColumns,
+ filters,
+ parts).asInstanceOf[RDD[Row]]
+ }
+}
diff --git
a/spark-iotdb-connector/src/main/scala/org/apache/iotdb/sparkdb/Transformer.scala
b/spark-iotdb-connector/src/main/scala/org/apache/iotdb/sparkdb/Transformer.scala
new file mode 100644
index 0000000..ae47fed
--- /dev/null
+++
b/spark-iotdb-connector/src/main/scala/org/apache/iotdb/sparkdb/Transformer.scala
@@ -0,0 +1,173 @@
+/**
+ * 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.sparkdb
+
+import org.apache.spark.sql.{SparkSession, _}
+import org.apache.spark.sql.types._
+
+
+object Transformer {
+ /**
+ * transfer old form to new form
+ *
+ * @param spark your SparkSession
+ * @param df dataFrame need to be transfer
+ *
+---------+-------------+-------------+-------------+-------------+-------------+-------------+
+ *
|timestamp|root.ln.d1.m1|root.ln.d1.m2|root.ln.d1.m3|root.ln.d2.m1|root.ln.d2.m2|root.ln.d2.m3|
+ *
+---------+-------------+-------------+-------------+-------------+-------------+-------------+
+ * | 1| 11| 12| null|
21| 22| 23|
+ *
+---------+-------------+-------------+-------------+-------------+-------------+-------------+
+ * @return tansferred data frame
+ * +---------+-----------+---+---+----+
+ * |timestamp|device_name| m1| m2| m3|
+ * +---------+-----------+---+---+----+
+ * | 1| root.ln.d2| 21| 22| 23|
+ * | 1| root.ln.d1| 11| 12|null|
+ * +---------+-----------+---+---+----+
+ */
+ def toNarrowForm(spark: SparkSession,
+ df: DataFrame): DataFrame = {
+ df.createOrReplaceTempView("tsfle_wide_form")
+ // use to record device and their measurement
+ var map = new scala.collection.mutable.HashMap[String, List[String]]()
+ // use to record all the measurement, prepare for the union
+ var m_map = scala.collection.mutable.HashMap[String, DataType]()
+
+ // this step is to record device_name and measurement_name
+ df.schema.foreach(f => {
+ if (!SQLConstant.TIMESTAMP_STR.equals(f.name)) {
+ val pos = f.name.lastIndexOf('.')
+ val divice_name = f.name.substring(0, pos)
+ val measurement_name = f.name.substring(pos + 1)
+ if (map.contains(divice_name)) {
+ map(divice_name) = map(divice_name) :+ measurement_name
+ }
+ else {
+ var l: List[String] = List()
+ l = l :+ measurement_name
+ map += (divice_name -> l)
+ }
+ m_map += (measurement_name -> f.dataType)
+ }
+ })
+
+ // we first get each device's measurement data and then union them to get
what we want, means:
+ // +---------+-----------+---+---+----+
+ // |timestamp|device_name| m1| m2| m3|
+ // +---------+-----------+---+---+----+
+ // | 1| root.ln.d2| 21| 22| 23|
+ // | 1| root.ln.d1| 11| 12|null|
+ // +---------+-----------+---+---+----+
+ var res: org.apache.spark.sql.DataFrame = null
+ map.keys.foreach { device_name =>
+ // build query
+ var query = "select " + SQLConstant.TIMESTAMP_STR + ", \"" + device_name
+ "\" as device_name"
+ val measurement_name = map(device_name)
+ m_map.keySet.foreach { m =>
+ val pos = measurement_name.indexOf(m)
+ if (pos >= 0) {
+ // select normal column
+ query += ", `" + device_name + "." + m + "` as " + m
+ }
+ else {
+ // fill null column
+ query += ", NULL as " + m
+ }
+ }
+
+ query += " from tsfle_wide_form"
+ var cur_df = spark.sql(query)
+
+ if (res == null) {
+ res = cur_df
+ }
+ else {
+ res = res.union(cur_df)
+ }
+ }
+
+ res
+ }
+
+ /**
+ * transfer new form to old form
+ *
+ * @param spark your SparkSession
+ * @param df dataFrame need to be tansfer
+ * +---------+-----------+---+---+----+
+ * |timestamp|device_name| m1| m2| m3|
+ * +---------+-----------+---+---+----+
+ * | 1| root.ln.d2| 21| 22| 23|
+ * | 1| root.ln.d1| 11| 12|null|
+ * +---------+-----------+---+---+----+
+ * @return tansferred data frame
+ *
+---------+-------------+-------------+-------------+-------------+-------------+-------------+
+ *
|timestamp|root.ln.d1.m1|root.ln.d1.m2|root.ln.d1.m3|root.ln.d2.m1|root.ln.d2.m2|root.ln.d2.m3|
+ *
+---------+-------------+-------------+-------------+-------------+-------------+-------------+
+ * | 1| 11| 12| null|
21| 22| 23|
+ *
+---------+-------------+-------------+-------------+-------------+-------------+-------------+
+ *
+ */
+ def toWideForm(spark: SparkSession,
+ df: DataFrame): DataFrame = {
+ df.createOrReplaceTempView("tsfle_narrow_form")
+ // get all device_name
+ val device_names = spark.sql("select distinct device_name from
tsfle_narrow_form").collect()
+ val table_df = spark.sql("select * from tsfle_narrow_form")
+
+ import scala.collection.mutable.ListBuffer
+ // get all measurement_name
+ val measurement_names = new ListBuffer[String]()
+
+ table_df.schema.foreach(f => {
+ if (!SQLConstant.TIMESTAMP_STR.equals(f.name) &&
!"device_name".equals(f.name)) {
+ measurement_names += f.name
+ }
+ })
+
+ var res: org.apache.spark.sql.DataFrame = null
+ // we first get each device's data and then join them with timestamp to
build tsRecord form which means:
+ //
+---------+-------------+-------------+-------------+-------------+-------------+-------------+
+ //
|timestamp|root.ln.d1.m1|root.ln.d1.m2|root.ln.d1.m3|root.ln.d2.m1|root.ln.d2.m2|root.ln.d2.m3|
+ //
+---------+-------------+-------------+-------------+-------------+-------------+-------------+
+ // | 1| 11| 12| null| 21|
22| 23|
+ //
+---------+-------------+-------------+-------------+-------------+-------------+-------------+
+
+ device_names.foreach(device_name => {
+ var query = "select " + SQLConstant.TIMESTAMP_STR
+
+ measurement_names.foreach(measurement_name => {
+ query = query + ", " + measurement_name + " as `" + device_name(0) +
"." + measurement_name + "`"
+ })
+
+ query = query + " from tsfle_narrow_form where device_name = \"" +
device_name(0) + "\""
+ val cur_df = spark.sql(query)
+
+ if (res == null) {
+ res = cur_df
+ }
+ else {
+ res = res.join(cur_df, List(SQLConstant.TIMESTAMP_STR), "outer")
+ }
+ })
+
+ res
+ }
+}
+
diff --git
a/spark-iotdb-connector/src/main/scala/org/apache/iotdb/sparkdb/package.scala
b/spark-iotdb-connector/src/main/scala/org/apache/iotdb/sparkdb/package.scala
new file mode 100644
index 0000000..7d946e1
--- /dev/null
+++
b/spark-iotdb-connector/src/main/scala/org/apache/iotdb/sparkdb/package.scala
@@ -0,0 +1,35 @@
+/**
+ * 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.sparkdb
+
+import org.apache.spark.sql.{DataFrame, DataFrameReader}
+
+package object sparkdb {
+
+ val myPackage = "org.apache.iotdb.sparkdb"
+
+ /**
+ * Adds a method, `iotdb`, to DataFrameReader that allows you to read data
from IoTDB using
+ * the DataFileReade
+ */
+ implicit class IoTDBDataFrameReader(reader: DataFrameReader) {
+ def iotdb: (Map[String, String]) => DataFrame =
reader.format(myPackage).options(_).load()
+ }
+
+}
\ No newline at end of file
diff --git
a/spark-iotdb-connector/src/test/scala/org/apache/iotdb/sparkdb/EnvironmentUtils.java
b/spark-iotdb-connector/src/test/scala/org/apache/iotdb/sparkdb/EnvironmentUtils.java
new file mode 100644
index 0000000..9e1f7de
--- /dev/null
+++
b/spark-iotdb-connector/src/test/scala/org/apache/iotdb/sparkdb/EnvironmentUtils.java
@@ -0,0 +1,282 @@
+/**
+ * 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.sparkdb;
+
+import java.io.File;
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Locale;
+import org.apache.commons.io.FileUtils;
+import org.apache.iotdb.db.auth.AuthException;
+import org.apache.iotdb.db.auth.authorizer.IAuthorizer;
+import org.apache.iotdb.db.auth.authorizer.LocalFileAuthorizer;
+import org.apache.iotdb.db.conf.IoTDBConfig;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.conf.adapter.IoTDBConfigDynamicAdapter;
+import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.StorageEngine;
+import org.apache.iotdb.db.engine.cache.DeviceMetaDataCache;
+import org.apache.iotdb.db.engine.cache.TsFileMetaDataCache;
+import org.apache.iotdb.db.engine.flush.FlushManager;
+import org.apache.iotdb.db.exception.StartupException;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.metadata.MManager;
+import org.apache.iotdb.db.monitor.StatMonitor;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.control.FileReaderManager;
+import org.apache.iotdb.db.query.control.QueryResourceManager;
+import org.apache.iotdb.db.writelog.manager.MultiFileLogNodeManager;
+import org.apache.iotdb.jdbc.Config;
+import org.junit.Assert;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * <p>
+ * This class is used for cleaning test environment in unit test and
integration test
+ * </p>
+ *
+ * @author liukun
+ *
+ */
+public class EnvironmentUtils {
+ private static String[] creationSqls = new String[]{
+ "SET STORAGE GROUP TO root.vehicle.d0",
+ "SET STORAGE GROUP TO root.vehicle.d1",
+ "CREATE TIMESERIES root.vehicle.d0.s0 WITH DATATYPE=INT32, ENCODING=RLE",
+ "CREATE TIMESERIES root.vehicle.d0.s1 WITH DATATYPE=INT64, ENCODING=RLE",
+ "CREATE TIMESERIES root.vehicle.d0.s2 WITH DATATYPE=FLOAT, ENCODING=RLE",
+ "CREATE TIMESERIES root.vehicle.d0.s3 WITH DATATYPE=TEXT,
ENCODING=PLAIN",
+ "CREATE TIMESERIES root.vehicle.d0.s4 WITH DATATYPE=BOOLEAN,
ENCODING=PLAIN"
+ };
+
+ private static String[] dataSet2 = new String[]{
+ "SET STORAGE GROUP TO root.ln.wf01.wt01",
+ "CREATE TIMESERIES root.ln.wf01.wt01.status WITH DATATYPE=BOOLEAN,
ENCODING=PLAIN",
+ "CREATE TIMESERIES root.ln.wf01.wt01.temperature WITH DATATYPE=FLOAT,
ENCODING=PLAIN",
+ "CREATE TIMESERIES root.ln.wf01.wt01.hardware WITH DATATYPE=INT32,
ENCODING=PLAIN",
+ "INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
+ + "values(1, 1.1, false, 11)",
+ "INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
+ + "values(2, 2.2, true, 22)",
+ "INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
+ + "values(3, 3.3, false, 33 )",
+ "INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
+ + "values(4, 4.4, false, 44)",
+ "INSERT INTO root.ln.wf01.wt01(timestamp,temperature,status, hardware) "
+ + "values(5, 5.5, false, 55)"
+ };
+
+ private static String insertTemplate = "INSERT INTO
root.vehicle.d0(timestamp,s0,s1,s2,s3,s4)"
+ + " VALUES(%d,%d,%d,%f,%s,%s)";
+
+ private static final Logger logger =
LoggerFactory.getLogger(EnvironmentUtils.class);
+
+ private static IoTDBConfig config =
IoTDBDescriptor.getInstance().getConfig();
+ private static DirectoryManager directoryManager =
DirectoryManager.getInstance();
+
+ public static long TEST_QUERY_JOB_ID =
QueryResourceManager.getInstance().assignJobId();
+ public static QueryContext TEST_QUERY_CONTEXT = new
QueryContext(TEST_QUERY_JOB_ID);
+
+ private static long oldTsFileThreshold = config.getTsFileSizeThreshold();
+
+ private static int oldMaxMemTableNumber = config.getMaxMemtableNumber();
+
+ private static long oldGroupSizeInByte = config.getMemtableSizeThreshold();
+
+ public static void cleanEnv() throws IOException, StorageEngineException {
+
+ QueryResourceManager.getInstance().endQueryForGivenJob(TEST_QUERY_JOB_ID);
+
+ // clear opened file streams
+ FileReaderManager.getInstance().closeAndRemoveAllOpenedReaders();
+
+ // clean storage group manager
+ if (!StorageEngine.getInstance().deleteAll()) {
+ logger.error("Can't close the storage group manager in
EnvironmentUtils");
+ Assert.fail();
+ }
+ StorageEngine.getInstance().reset();
+ IoTDBDescriptor.getInstance().getConfig().setReadOnly(false);
+
+ StatMonitor.getInstance().close();
+ // clean wal
+ MultiFileLogNodeManager.getInstance().stop();
+ // clean cache
+ if (config.isMetaDataCacheEnable()) {
+ TsFileMetaDataCache.getInstance().clear();
+ DeviceMetaDataCache.getInstance().clear();
+ }
+ // close metadata
+ MManager.getInstance().clear();
+ // delete all directory
+ cleanAllDir();
+
+ config.setMaxMemtableNumber(oldMaxMemTableNumber);
+ config.setTsFileSizeThreshold(oldTsFileThreshold);
+ config.setMemtableSizeThreshold(oldGroupSizeInByte);
+ IoTDBConfigDynamicAdapter.getInstance().reset();
+ }
+
+ public static void cleanAllDir() throws IOException {
+ // delete sequential files
+ for (String path : directoryManager.getAllSequenceFileFolders()) {
+ cleanDir(path);
+ }
+ // delete unsequence files
+ for (String path : directoryManager.getAllUnSequenceFileFolders()) {
+ cleanDir(path);
+ }
+ // delete system info
+ cleanDir(config.getSystemDir());
+ // delete wal
+ cleanDir(config.getWalFolder());
+ // delete index
+ cleanDir(config.getIndexFileDir());
+ cleanDir(config.getBaseDir());
+ // delete data files
+ for (String dataDir : config.getDataDirs()) {
+ cleanDir(dataDir);
+ }
+ }
+
+ public static void cleanDir(String dir) throws IOException {
+ FileUtils.deleteDirectory(new File(dir));
+ }
+
+ /**
+ * disable the system monitor</br>
+ * this function should be called before all code in the setup
+ */
+ public static void closeStatMonitor() {
+ config.setEnableStatMonitor(false);
+ }
+
+ /**
+ * disable memory control</br>
+ * this function should be called before all code in the setup
+ */
+ public static void envSetUp() throws StartupException, IOException {
+ IoTDBDescriptor.getInstance().getConfig().setEnableParameterAdapter(false);
+ MManager.getInstance().init();
+ IoTDBConfigDynamicAdapter.getInstance().setInitialized(true);
+
+ createAllDir();
+ // disable the system monitor
+ config.setEnableStatMonitor(false);
+ IAuthorizer authorizer;
+ try {
+ authorizer = LocalFileAuthorizer.getInstance();
+ } catch (AuthException e) {
+ throw new StartupException(e);
+ }
+ try {
+ authorizer.reset();
+ } catch (AuthException e) {
+ throw new StartupException(e);
+ }
+ StorageEngine.getInstance().reset();
+ MultiFileLogNodeManager.getInstance().start();
+ FlushManager.getInstance().start();
+ TEST_QUERY_JOB_ID = QueryResourceManager.getInstance().assignJobId();
+ TEST_QUERY_CONTEXT = new QueryContext(TEST_QUERY_JOB_ID);
+ }
+
+ private static void createAllDir() {
+ // create sequential files
+ for (String path : directoryManager.getAllSequenceFileFolders()) {
+ createDir(path);
+ }
+ // create unsequential files
+ for (String path : directoryManager.getAllUnSequenceFileFolders()) {
+ createDir(path);
+ }
+ // create storage group
+ createDir(config.getSystemDir());
+ // create wal
+ createDir(config.getWalFolder());
+ // create index
+ createDir(config.getIndexFileDir());
+ // create data
+ for (String dataDir: config.getDataDirs()) {
+ createDir(dataDir);
+ }
+ }
+
+ private static void createDir(String dir) {
+ File file = new File(dir);
+ file.mkdirs();
+ }
+
+ public static void prepareData() throws SQLException {
+ try (Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root",
+ "root");
+ Statement statement = connection.createStatement()) {
+
+ for (String sql : creationSqls) {
+ statement.execute(sql);
+ }
+
+ for (String sql : dataSet2) {
+ statement.execute(sql);
+ }
+
+ // prepare BufferWrite file
+ for (int i = 5000; i < 7000; i++) {
+ statement.execute(String
+ .format(Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "\'"
+ i + "\'", "true"));
+ }
+ statement.execute("flush");
+ for (int i = 7500; i < 8500; i++) {
+ statement.execute(String
+ .format(Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "\'"
+ i + "\'", "false"));
+ }
+ statement.execute("flush");
+ // prepare Unseq-File
+ for (int i = 500; i < 1500; i++) {
+ statement.execute(String
+ .format(Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "\'"
+ i + "\'", "true"));
+ }
+ statement.execute("flush");
+ for (int i = 3000; i < 6500; i++) {
+ statement.execute(String
+ .format(Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "\'"
+ i + "\'", "false"));
+ }
+ statement.execute("merge");
+
+ // prepare BufferWrite cache
+ for (int i = 9000; i < 10000; i++) {
+ statement.execute(String
+ .format(Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "\'"
+ i + "\'", "true"));
+ }
+ // prepare Overflow cache
+ for (int i = 2000; i < 2500; i++) {
+ statement.execute(String
+ .format(Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "\'"
+ i + "\'", "false"));
+ }
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+}
diff --git
a/spark-iotdb-connector/src/test/scala/org/apache/iotdb/sparkdb/IoTDBTest.scala
b/spark-iotdb-connector/src/test/scala/org/apache/iotdb/sparkdb/IoTDBTest.scala
new file mode 100644
index 0000000..e4e9eaa
--- /dev/null
+++
b/spark-iotdb-connector/src/test/scala/org/apache/iotdb/sparkdb/IoTDBTest.scala
@@ -0,0 +1,115 @@
+/**
+ * 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.sparkdb
+
+import org.apache.iotdb.db.service.IoTDB
+import org.apache.iotdb.jdbc.Config
+import org.apache.spark.sql.{SQLContext, SparkSession}
+import org.junit._
+import org.scalatest.{BeforeAndAfterAll, FunSuite}
+
+class IoTDBTest extends FunSuite with BeforeAndAfterAll {
+ private var daemon: IoTDB = _
+
+ private val testFile =
"/home/hadoop/git/tsfile/delta-spark/src/test/resources/test.tsfile"
+ private val csvPath: java.lang.String =
"/home/hadoop/git/tsfile/delta-spark/src/test/resources/test.csv"
+ private val tsfilePath: java.lang.String =
"/home/hadoop/git/tsfile/delta-spark/src/test/resources/test.tsfile"
+ private val errorPath: java.lang.String =
"/home/hadoop/git/tsfile/delta-spark/src/test/resources/errortest.tsfile"
+ private var sqlContext: SQLContext = _
+ private var spark: SparkSession = _
+
+ @Before
+ override protected def beforeAll(): Unit = {
+ super.beforeAll()
+
+ EnvironmentUtils.closeStatMonitor()
+ daemon = IoTDB.getInstance
+ daemon.active()
+ EnvironmentUtils.envSetUp()
+ Class.forName(Config.JDBC_DRIVER_NAME)
+ EnvironmentUtils.prepareData()
+
+ spark = SparkSession
+ .builder()
+ .config("spark.master", "local")
+ .appName("TSFile test")
+ .getOrCreate()
+ }
+
+ @AfterClass
+ override protected def afterAll(): Unit = {
+ if (spark != null) {
+ spark.sparkContext.stop()
+ }
+
+ daemon.stop()
+ EnvironmentUtils.cleanEnv()
+
+ super.afterAll()
+ }
+
+ test("test show data") {
+ val df = spark.read.format("org.apache.iotdb.sparkdb").option("url",
"jdbc:iotdb://127.0.0.1:6667/").option("sql", "select * from root").load
+ df.printSchema()
+ df.show()
+ Assert.assertEquals(7505, df.count())
+ }
+
+ test("test show data with partition") {
+ val df = spark.read.format("org.apache.iotdb.sparkdb").option("url",
"jdbc:iotdb://127.0.0.1:6667/").option("sql", "select * from
root").option("lowerBound", 1).option("upperBound", System.nanoTime() / 1000 /
1000).option("numPartition", 10).load
+
+ df.printSchema()
+
+ df.show()
+
+ Assert.assertEquals(7505, df.count())
+ }
+
+ test("test filter data") {
+ val df = spark.read.format("org.apache.iotdb.sparkdb").option("url",
"jdbc:iotdb://127.0.0.1:6667/").option("sql", "select * from root where time <
2000 and time > 1000").load
+
+ Assert.assertEquals(499, df.count())
+ }
+
+ test("test filter data with partition") {
+ val df = spark.read.format("org.apache.iotdb.sparkdb").option("url",
"jdbc:iotdb://127.0.0.1:6667/").option("sql", "select * from root where time <
2000 and time > 1000").option("lowerBound", 1).option("upperBound",
10000).option("numPartition", 10).load
+
+ Assert.assertEquals(499, df.count())
+ }
+
+ test("test transform to narrow") {
+ val df = spark.read.format("org.apache.iotdb.sparkdb").option("url",
"jdbc:iotdb://127.0.0.1:6667/").option("sql", "select * from root where time <
1100 and time > 1000").load
+ val narrow_df = Transformer.toNarrowForm(spark, df)
+ Assert.assertEquals(198, narrow_df.count())
+ }
+
+ test("test transform to narrow with partition") {
+ val df = spark.read.format("org.apache.iotdb.sparkdb").option("url",
"jdbc:iotdb://127.0.0.1:6667/").option("sql", "select * from root where time <
1100 and time > 1000").option("lowerBound", 1).option("upperBound",
10000).option("numPartition", 10).load
+ val narrow_df = Transformer.toNarrowForm(spark, df)
+ Assert.assertEquals(198, narrow_df.count())
+ }
+
+ test("test transform back to wide") {
+ val df = spark.read.format("org.apache.iotdb.sparkdb").option("url",
"jdbc:iotdb://127.0.0.1:6667/").option("sql", "select * from root where time <
1100 and time > 1000").load
+ val narrow_df = Transformer.toNarrowForm(spark, df)
+ val wide_df = Transformer.toWideForm(spark, narrow_df)
+ Assert.assertEquals(99, wide_df.count())
+ }
+}
+
