Apache Spark

版本支持

支持的 Spark 与 Scala 版本如下：

注意事项

1. 当前版本的 spark-iotdb-connector 支持 2.11 与 2.12 两个版本的 Scala，暂不支持 2.13 版本。
2. spark-iotdb-connector 支持在 Java、Scala 版本的 Spark 与 PySpark 中使用。

部署

spark-iotdb-connector 总共有两个使用场景，分别为 IDE 开发与 spark-shell 调试。

IDE 开发

在 IDE 开发时，只需要在 pom.xml 文件中添加以下依赖即可：

1. <dependency>
2. <groupId>org.apache.iotdb</groupId>
3. <!-- spark-iotdb-connector_2.11 or spark-iotdb-connector_2.13 -->
4. <artifactId>spark-iotdb-connector_2.12.10</artifactId>
5. <version>${iotdb.version}</version>
6. </dependency>

spark-shell 调试

如果需要在 spark-shell 中使用 spark-iotdb-connetcor，需要先在官网下载 with-dependencies 版本的 jar 包。然后再将 Jar 包拷贝到 ${SPARK_HOME}/jars 目录中即可。 执行以下命令即可：

1. cp spark-iotdb-connector_2.12.10-${iotdb.version}.jar $SPARK_HOME/jars/

此外，为了保证 spark 能使用 JDBC 和 IoTDB 连接，需要进行如下操作:

运行如下命令来编译 IoTDB-JDBC 连接器：

1. mvn clean package -pl iotdb-client/jdbc -am -DskipTests -P get-jar-with-dependencies

编译后的 jar 包在如下目录中：

1. $IoTDB_HOME/iotdb-client/jdbc/target/iotdb-jdbc-{version}-SNAPSHOT-jar-with-dependencies.jar

最后再将 jar 包拷贝到 ${SPARK_HOME}/jars 目录中即可。执行以下命令即可：

1. cp iotdb-jdbc-{version}-SNAPSHOT-jar-with-dependencies.jar $SPARK_HOME/jars/

使用

参数

从 IoTDB 读取数据

以下是一个示例，演示如何从 IoTDB 中读取数据成为 DataFrame。

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

3. val df = spark.read.format("org.apache.iotdb.spark.db")
4. .option("user", "root")
5. .option("password", "root")
6. .option("url", "jdbc:iotdb://127.0.0.1:6667/")
7. .option("sql", "select ** from root") // 查询 SQL
8. .option("lowerBound", "0") // 时间戳下界
9. .option("upperBound", "100000000") // 时间戳上界
10. .option("numPartition", "5") // 分区数
11. .load

13. df.printSchema()

15. df.show()

将数据写入 IoTDB

以下是一个示例，演示如何将数据写入 IoTDB。

1. // 构造窄表数据
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")
8. .withColumnRenamed("_3", "s0")
9. .withColumnRenamed("_4", "s1")
10. .withColumnRenamed("_5", "s2")
11. .withColumnRenamed("_6", "s3")
12. .withColumnRenamed("_7", "s4")
13. .withColumnRenamed("_8", "s5")

15. // 写入窄表数据
16. dfWithColumn
17. .write
18. .format("org.apache.iotdb.spark.db")
19. .option("url", "jdbc:iotdb://127.0.0.1:6667/")
20. .save

22. // 构造宽表数据
23. val df = spark.createDataFrame(List(
24. (1L, 1, 1L, 1.0F, 1.0D, true, "hello"),
25. (2L, 2, 2L, 2.0F, 2.0D, false, "world")))

27. val dfWithColumn = df.withColumnRenamed("_1", "Time")
28. .withColumnRenamed("_2", "root.test.d0.s0")
29. .withColumnRenamed("_3", "root.test.d0.s1")
30. .withColumnRenamed("_4", "root.test.d0.s2")
31. .withColumnRenamed("_5", "root.test.d0.s3")
32. .withColumnRenamed("_6", "root.test.d0.s4")
33. .withColumnRenamed("_7", "root.test.d0.s5")

35. // 写入宽表数据
36. dfWithColumn.write.format("org.apache.iotdb.spark.db")
37. .option("url", "jdbc:iotdb://127.0.0.1:6667/")
38. .option("numPartition", "10")
39. .save

宽表与窄表转换

以下是如何转换宽表与窄表的示例：

• 从宽到窄

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)

宽表与窄表

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

TsFile 中的现有数据如下：

• d1:root.ln.wf01.wt01
• d2:root.ln.wf02.wt02

宽（默认）表形式如下：

你还可以使用窄表形式，如下所示：