Spark IoTDB连接器

版本

Spark和Java所需的版本如下：

Spark Version	Scala Version	Java Version	TsFile
`2.4.3`	`2.11`	`1.8`	`0.11.1`

安装

mvn clean scala:compile compile install

1. Maven依赖

    <dependency>
      <groupId>org.apache.iotdb</groupId>
      <artifactId>spark-iotdb-connector</artifactId>
      <version>0.11.1</version>
    </dependency>

2. Spark-shell用户指南

spark-shell --jars spark-iotdb-connector-0.11.1.jar,iotdb-jdbc-0.11.1-jar-with-dependencies.jar
import org.apache.iotdb.spark.db._
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
df.printSchema()
df.show()

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

spark-shell --jars spark-iotdb-connector-0.11.1.jar,iotdb-jdbc-0.11.1-jar-with-dependencies.jar
import org.apache.iotdb.spark.db._
val df = spark.read.format("org.apache.iotdb.spark.db").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. 模式推断

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

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

TsFile中的现有数据如下：

device:root.ln.wf01.wt01				device:root.ln.wf02.wt02
状态		温度		硬件		状态
时间	值	时间	值	时间	值	时间	值
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

4. 在宽和窄表之间转换

从宽到窄

import org.apache.iotdb.spark.db._
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
val narrow_df = Transformer.toNarrowForm(spark, wide_df)

从窄到宽

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

5. Java用户指南

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.iotdb.spark.db.*
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.spark.db")
        .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()
  }
}