Spark-IoTDB

版本

Spark和Java所需的版本如下：

Spark Version Scala Version Java Version TsFile
2.4.0-3.2.0 2.12 1.8 0.13.0

注意

  1. Spark IoTDB Connector只支持Spark 2.4.0到Spark 3.2.0的Scala 2.12版本。 如果需要对其他版本进行支持，可以通过修改源码中spark-iotdb-connector这个模块里面pom文件的Scala版本之后进行重新编译。
  2. 因为IoTDB与Spark的thrift版本有冲突，所以需要通过执行rm -f $SPARK_HOME/jars/libthrift*和cp $IOTDB_HOME/lib/libthrift* $SPARK_HOME/jars/这两个命令来解决。 否则的话，就只能在IDE里面进行代码调试。而且如果你需要通过spark-submit命令提交任务的话，你打包时必须要带上依赖。

安装

mvn clean scala:compile compile install

Maven依赖

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

Spark-shell用户指南

1. spark-shell --jars spark-iotdb-connector-0.13.0.jar,iotdb-jdbc-0.13.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()

如果要对rdd进行分区，可以执行以下操作

1. spark-shell --jars spark-iotdb-connector-0.13.0.jar,iotdb-jdbc-0.13.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()

模式推断

以下TsFile结构为例：TsFile模式中有三个度量：状态，温度和硬件。 这三种测量的基本信息如下：

名称 类型 编码
状态 Boolean PLAIN
温度 Float RLE
硬件 Text PLAIN

TsFile中的现有数据如下：

  • d1:root.ln.wf01.wt01
  • d2:root.ln.wf02.wt02
time d1.status time d1.temperature time d2.hardware time d2.status
1 True 1 2.2 2 “aaa” 1 True
3 True 2 2.2 4 “bbb” 2 False
5 False 3 2.1 6 “ccc” 4 True

宽（默认）表形式如下：

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

你还可以使用窄表形式，如下所示：（您可以参阅第4部分，了解如何使用窄表形式）

时间 设备名 状态 硬件 温度
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

在宽和窄表之间转换

  • 从宽到窄
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)
  • 从窄到宽
1. import org.apache.iotdb.spark.db._

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

Java用户指南

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

写数据到IoTDB

用户指南

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

注意

  1. 无论dataframe中存放的是窄表还是宽表，都可以直接将数据写到IoTDB中。
  2. numPartition参数是用来设置分区数，会在写入数据之前给dataframe进行重分区。每一个分区都会开启一个session进行数据的写入，来提高并发数。