Ecosystem Integration - Spark IoTDB - 《Apache IoTDB User Guide (V0.12.x)》

Spark-IoTDB
- version
- Install
Write Data to IoTDB
- User Guide
- Notes

Spark-IoTDB

version

The versions required for Spark and Java are as follow:

Spark Version	Scala Version	Java Version	TsFile
`2.4.5`	`2.12`	`1.8`	`0.12.0`

Currently we only support spark version 2.4.3 and there are some known issue on 2.4.7, do no use it

Install

mvn clean scala:compile compile install

Maven Dependency

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

spark-shell user guide

1. spark-shell --jars spark-iotdb-connector-0.12.0.jar,iotdb-jdbc-0.12.0-jar-with-dependencies.jar

3. import org.apache.iotdb.spark.db._

5. 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

7. df.printSchema()

9. df.show()

To partition rdd:

1. spark-shell --jars spark-iotdb-connector-0.12.0.jar,iotdb-jdbc-0.12.0-jar-with-dependencies.jar

3. import org.apache.iotdb.spark.db._

5. val df = spark.read.format("org.apache.iotdb.spark.db").option("url","jdbc:iotdb://127.0.0.1:6667/").option("sql","select * from root").
6. option("lowerBound", [lower bound of time that you want query(include)]).option("upperBound", [upper bound of time that you want query(include)]).
7. option("numPartition", [the partition number you want]).load

9. df.printSchema()

11. df.show()

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:

Name	Type	Encode
status	Boolean	PLAIN
temperature	Float	RLE
hardware	Text	PLAIN

The existing data in the TsFile is as follows:

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

Transform between wide and narrow table

from wide to narrow

1. import org.apache.iotdb.spark.db._

3. 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
4. val narrow_df = Transformer.toNarrowForm(spark, wide_df)

from narrow to wide

1. import org.apache.iotdb.spark.db._

3. val wide_df = Transformer.toWideForm(spark, narrow_df)

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.*

6. public class Example {

8. public static void main(String[] args) {
9. SparkSession spark = SparkSession
10. .builder()
11. .appName("Build a DataFrame from Scratch")
12. .master("local[*]")
13. .getOrCreate();

15. Dataset<Row> df = spark.read().format("org.apache.iotdb.spark.db")
16. .option("url","jdbc:iotdb://127.0.0.1:6667/")
17. .option("sql","select * from root").load();

19. df.printSchema();

21. df.show();

23. Dataset<Row> narrowTable = Transformer.toNarrowForm(spark, df)
24. narrowTable.show()
25. }
26. }

Write Data to IoTDB

User Guide

1. // import narrow table
2. val df = spark.createDataFrame(List(
3. (1L, "root.test.d0",1, 1L, 1.0F, 1.0D, true, "hello"),
4. (2L, "root.test.d0", 2, 2L, 2.0F, 2.0D, false, "world")))

6. val dfWithColumn = df.withColumnRenamed("_1", "Time")
7. .withColumnRenamed("_2", "device_name")
8. .withColumnRenamed("_3", "s0")
9. .withColumnRenamed("_4", "s1")
10. .withColumnRenamed("_5", "s2")
11. .withColumnRenamed("_6", "s3")
12. .withColumnRenamed("_7", "s4")
13. .withColumnRenamed("_8", "s5")
14. dfWithColumn
15. .write
16. .format("org.apache.iotdb.spark.db")
17. .option("url", "jdbc:iotdb://127.0.0.1:6667/")
18. .save

20. // import wide table
21. val df = spark.createDataFrame(List(
22. (1L, 1, 1L, 1.0F, 1.0D, true, "hello"),
23. (2L, 2, 2L, 2.0F, 2.0D, false, "world")))

25. val dfWithColumn = df.withColumnRenamed("_1", "Time")
26. .withColumnRenamed("_2", "root.test.d0.s0")
27. .withColumnRenamed("_3", "root.test.d0.s1")
28. .withColumnRenamed("_4", "root.test.d0.s2")
29. .withColumnRenamed("_5", "root.test.d0.s3")
30. .withColumnRenamed("_6", "root.test.d0.s4")
31. .withColumnRenamed("_7", "root.test.d0.s5")
32. dfWithColumn.write.format("org.apache.iotdb.spark.db")
33. .option("url", "jdbc:iotdb://127.0.0.1:6667/")
34. .option("numPartition", "10")
35. .save

Notes

You can directly write data to IoTDB whatever the dataframe contains a wide table or a narrow table.
The parameter numPartition is used to set the number of partitions. The dataframe that you want to save will be repartition based on this parameter before writing data. Each partition will open a session to write data to increase the number of concurrent requests.