Spark IoTDB Connecter

version

The versions required for Spark and Java are as follow:

Spark VersionScala VersionJava VersionTsFile
2.4.32.111.80.10.0

install

mvn clean scala:compile compile install

1. maven dependency

  1. <dependency>
  2. <groupId>org.apache.iotdb</groupId>
  3. <artifactId>spark-iotdb-connector</artifactId>
  4. <version>0.10.0</version>
  5. </dependency>

2. spark-shell user guide

  1. spark-shell --jars spark-iotdb-connector-0.10.0.jar,iotdb-jdbc-0.10.0-jar-with-dependencies.jar
  2. import org.apache.iotdb.spark.db._
  3. val df = spark.read.format("org.apache.iotdb.spark.db").option("url","jdbc:iotdb://127.0.0.1:6667/").option("sql","select * from root").load
  4. df.printSchema()
  5. df.show()

if you want to partition your rdd, you can do as following

  1. spark-shell --jars spark-iotdb-connector-0.10.0.jar,iotdb-jdbc-0.10.0-jar-with-dependencies.jar
  2. import org.apache.iotdb.spark.db._
  3. val df = spark.read.format("org.apache.iotdb.spark.db").option("url","jdbc:iotdb://127.0.0.1:6667/").option("sql","select * from root").
  4. option("lowerBound", [lower bound of time that you want query(include)]).option("upperBound", [upper bound of time that you want query(include)]).
  5. option("numPartition", [the partition number you want]).load
  6. df.printSchema()
  7. 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:

NameTypeEncode
statusBooleanPLAIN
temperatureFloatRLE
hardwareTextPLAIN

The existing data in the TsFile is as follows:

device:root.ln.wf01.wt01device:root.ln.wf02.wt02
statustemperaturehardwarestatus
timevaluetimevaluetimevaluetimevalue
1True12.22“aaa”1True
3True22.24“bbb”2False
5False32.16“ccc”4True

The wide(default) table form is as follows:

timeroot.ln.wf02.wt02.temperatureroot.ln.wf02.wt02.statusroot.ln.wf02.wt02.hardwareroot.ln.wf01.wt01.temperatureroot.ln.wf01.wt01.statusroot.ln.wf01.wt01.hardware
1nulltruenull2.2truenull
2nullfalseaaa2.2nullnull
3nullnullnull2.1truenull
4nulltruebbbnullnullnull
5nullnullnullnullfalsenull
6nullnullcccnullnullnull

You can also use narrow table form which as follows: (You can see part 4 about how to use narrow form)

timedevice_namestatushardwaretemperature
1root.ln.wf02.wt01truenull2.2
1root.ln.wf02.wt02truenullnull
2root.ln.wf02.wt01nullnull2.2
2root.ln.wf02.wt02falseaaanull
3root.ln.wf02.wt01truenull2.1
4root.ln.wf02.wt02truebbbnull
5root.ln.wf02.wt01falsenullnull
6root.ln.wf02.wt02nullcccnull

4. Transform between wide and narrow table

from wide to narrow

  1. import org.apache.iotdb.spark.db._
  2. val wide_df = spark.read.format("org.apache.iotdb.spark.db").option("url", "jdbc:iotdb://127.0.0.1:6667/").option("sql", "select * from root where time < 1100 and time > 1000").load
  3. val narrow_df = Transformer.toNarrowForm(spark, wide_df)

from narrow to wide

  1. import org.apache.iotdb.spark.db._
  2. val wide_df = Transformer.toWideForm(spark, narrow_df)

5. Java user guide

  1. import org.apache.spark.sql.Dataset;
  2. import org.apache.spark.sql.Row;
  3. import org.apache.spark.sql.SparkSession;
  4. import org.apache.iotdb.spark.db.*
  5. public class Example {
  6. public static void main(String[] args) {
  7. SparkSession spark = SparkSession
  8. .builder()
  9. .appName("Build a DataFrame from Scratch")
  10. .master("local[*]")
  11. .getOrCreate();
  12. Dataset<Row> df = spark.read().format("org.apache.iotdb.spark.db")
  13. .option("url","jdbc:iotdb://127.0.0.1:6667/")
  14. .option("sql","select * from root").load();
  15. df.printSchema();
  16. df.show();
  17. Dataset<Row> narrowTable = Transformer.toNarrowForm(spark, df)
  18. narrowTable.show()
  19. }
  20. }