Apache IoTDB开发系统之Java 原生接口

安装

依赖

• JDK >= 1.8
• Maven >= 3.6

安装方法

在根目录下运行：

1. mvn clean install -pl session -am -Dmaven.test.skip=true

在 MAVEN 中使用原生接口

1. <dependencies>
2. <dependency>
3. <groupId>org.apache.iotdb</groupId>
4. <artifactId>iotdb-session</artifactId>
5. <version>0.13.0-SNAPSHOT</version>
6. </dependency>
7. </dependencies>

语法说明

• 对于 IoTDB-SQL 接口：传入的 SQL 参数需要符合 语法规范 ，并且针对 JAVA 字符串进行反转义，如双引号前需要加反斜杠。（即：经 JAVA 转义之后与命令行执行的 SQL 语句一致。）
• 对于其他接口：
  • 经参数传入的路径或路径前缀中的节点：
    • 在 SQL 语句中需要使用反引号（`）进行转义的，此处均不需要进行转义。
    • 使用单引号或双引号括起的节点，仍需要使用单引号或双引号括起，并且要针对 JAVA 字符串进行反转义。
    • 对于 接口，由于内部调用了 IoTDB-SQL 接口，因此需要和 SQL 语法规范保持一致，并且针对 JAVA 字符串进行反转义。checkTimeseriesExists
  • 经参数传入的标识符（如模板名）：在 SQL 语句中需要使用反引号（`）进行转义的，此处均不需要进行转义。

基本接口说明

下面将给出 Session 对应的接口的简要介绍和对应参数：

初始化

• 初始化 Session

1. // 全部使用默认配置
2. session = new Session.Builder.build();
4. // 指定一个可连接节点
5. session =
6. new Session.Builder()
7. .host(String host)
8. .port(int port)
9. .build();
11. // 指定多个可连接节点
12. session =
13. new Session.Builder()
14. .nodeUrls(List<String> nodeUrls)
15. .build();
17. // 其他配置项
18. session =
19. new Session.Builder()
20. .fetchSize(int fetchSize)
21. .username(String username)
22. .password(String password)
23. .thriftDefaultBufferSize(int thriftDefaultBufferSize)
24. .thriftMaxFrameSize(int thriftMaxFrameSize)
25. .enableCacheLeader(boolean enableCacheLeader)
26. .version(Version version)
27. .build();

其中，version 表示客户端使用的 SQL 语义版本，用于升级 0.13 时兼容 0.12 的 SQL 语义，可能取值有：、。V_0_12V_0_13

• 开启 Session

1. void open()

• 开启 Session，并决定是否开启 RPC 压缩

1. void open(boolean enableRPCCompression)

注意: 客户端的 RPC 压缩开启状态需和服务端一致

• 关闭 Session

1. void close()

数据定义接口 DDL

存储组管理

• 设置存储组

1. void setStorageGroup(String storageGroupId)

• 删除单个或多个存储组

1. void deleteStorageGroup(String storageGroup)
2. void deleteStorageGroups(List<String> storageGroups)

时间序列管理

• 创建单个或多个时间序列

1. void createTimeseries(String path, TSDataType dataType,
2. TSEncoding encoding, CompressionType compressor, Map<String, String> props,
3. Map<String, String> tags, Map<String, String> attributes, String measurementAlias)
5. void createMultiTimeseries(List<String> paths, List<TSDataType> dataTypes,
6. List<TSEncoding> encodings, List<CompressionType> compressors,
7. List<Map<String, String>> propsList, List<Map<String, String>> tagsList,
8. List<Map<String, String>> attributesList, List<String> measurementAliasList)

• 创建对齐时间序列

1. void createAlignedTimeseries(String prefixPath, List<String> measurements,
2. List<TSDataType> dataTypes, List<TSEncoding> encodings,
3. CompressionType compressor, List<String> measurementAliasList);

注意：目前暂不支持使用传感器别名。

• 删除一个或多个时间序列

1. void deleteTimeseries(String path)
2. void deleteTimeseries(List<String> paths)

• 检测时间序列是否存在

1. boolean checkTimeseriesExists(String path)

元数据模版

• 创建元数据模板，可以通过先后创建、的对象，描述模板内物理量结构与类型、编码方式、压缩方式等信息，并通过以下接口创建模板TemplateMeasurementNode

1. public void createSchemaTemplate(Template template);
3. Class Template {
4. private String name;
5. private boolean directShareTime;
6. Map<String, Node> children;
7. public Template(String name, boolean isShareTime);
9. public void addToTemplate(Node node);
10. public void deleteFromTemplate(String name);
11. public void setShareTime(boolean shareTime);
12. }
14. Abstract Class Node {
15. private String name;
16. public void addChild(Node node);
17. public void deleteChild(Node node);
18. }
20. Class MeasurementNode extends Node {
21. TSDataType dataType;
22. TSEncoding encoding;
23. CompressionType compressor;
24. public MeasurementNode(String name,
25. TSDataType dataType,
26. TSEncoding encoding,
27. CompressionType compressor);
28. }

通过上述类的实例描述模板时，内应当仅能包含单层的，具体可以参见如下示例：TemplateMeasurementNode

1. MeasurementNode nodeX = new MeasurementNode("x", TSDataType.FLOAT, TSEncoding.RLE, CompressionType.SNAPPY);
2. MeasurementNode nodeY = new MeasurementNode("y", TSDataType.FLOAT, TSEncoding.RLE, CompressionType.SNAPPY);
3. MeasurementNode nodeSpeed = new MeasurementNode("speed", TSDataType.DOUBLE, TSEncoding.GORILLA, CompressionType.SNAPPY);
5. // This is the template we suggest to implement
6. Template flatTemplate = new Template("flatTemplate");
7. template.addToTemplate(nodeX);
8. template.addToTemplate(nodeY);
9. template.addToTemplate(nodeSpeed);
11. createSchemaTemplate(flatTemplate);

• 在创建概念元数据模板以后，还可以通过以下接口增加或删除模板内的物理量。请注意，已经挂载的模板不能删除内部的物理量。

1. // 为指定模板新增一组对齐的物理量，若其父节点在模板中已经存在，且不要求对齐，则报错
2. public void addAlignedMeasurementsInTemplate(String templateName,
3. String[] measurementsPath,
4. TSDataType[] dataTypes,
5. TSEncoding[] encodings,
6. CompressionType[] compressors);
8. // 为指定模板新增一个对齐物理量, 若其父节点在模板中已经存在，且不要求对齐，则报错
9. public void addAlignedMeasurementInTemplate(String templateName,
10. String measurementPath,
11. TSDataType dataType,
12. TSEncoding encoding,
13. CompressionType compressor);
16. // 为指定模板新增一个不对齐物理量, 若其父节在模板中已经存在，且要求对齐，则报错
17. public void addUnalignedMeasurementInTemplate(String templateName,
18. String measurementPath,
19. TSDataType dataType,
20. TSEncoding encoding,
21. CompressionType compressor);
23. // 为指定模板新增一组不对齐的物理量, 若其父节在模板中已经存在，且要求对齐，则报错
24. public void addUnalignedMeasurementsIntemplate(String templateName,
25. String[] measurementPaths,
26. TSDataType[] dataTypes,
27. TSEncoding[] encodings,
28. CompressionType[] compressors);
30. // 从指定模板中删除一个节点
31. public void deleteNodeInTemplate(String templateName, String path);

• 对于已经创建的元数据模板，还可以通过以下接口查询模板信息：

1. // 查询返回目前模板中所有物理量的数量
2. public int countMeasurementsInTemplate(String templateName);
4. // 检查模板内指定路径是否为物理量
5. public boolean isMeasurementInTemplate(String templateName, String path);
7. // 检查在指定模板内是否存在某路径
8. public boolean isPathExistInTemplate(String templateName, String path);
10. // 返回指定模板内所有物理量的路径
11. public List<String> showMeasurementsInTemplate(String templateName);
13. // 返回指定模板内某前缀路径下的所有物理量的路径
14. public List<String> showMeasurementsInTemplate(String templateName, String pattern);

• 将名为的元数据模板挂载到’prefixPath’路径下，在执行这一步之前，你需要创建名为的元数据模板templateNametemplateName
• 请注意，我们强烈建议您将模板设置在存储组或存储组下层的节点中，以更好地适配未来版本更新及各模块的协作

1. void setSchemaTemplate(String templateName, String prefixPath)

• 将模板挂载到 MTree 上之后，你可以随时查询所有模板的名称、某模板被设置到 MTree 的所有路径、所有正在使用某模板的所有路径，即如下接口：

1. /** @return All template names. */
2. public List<String> showAllTemplates();
4. /** @return All paths have been set to designated template. */
5. public List<String> showPathsTemplateSetOn(String templateName);
7. /** @return All paths are using designated template. */
8. public List<String> showPathsTemplateUsingOn(String templateName)

• 如果你需要删除某一个模板，请确保在进行删除之前，MTree 上已经没有节点被挂载了模板，对于已经被挂载模板的节点，可以用如下接口卸载模板；

1. void unsetSchemaTemplate(String prefixPath, String templateName);
2. public void dropSchemaTemplate(String templateName);

• 请注意，如果一个子树中有多个孩子节点需要使用模板，可以在其共同父母节点上使用 。而只有在已有数据点插入模板对应的物理量，或使用以下接口激活模板，模板才会被设置为激活状态，进而被 等查询访问到。setSchemaTemplateshow timeseries

1. public void createTimeseriesOfTemplateOnPath(String path);

• 卸载’prefixPath’路径下的名为的元数据模板。你需要保证给定的路径下需要有名为的元数据模板。templateNameprefixPathtemplateName
• 如果在挂载模板后，曾经在路径及其后代节点使用模板插入数据后，或者使用了激活模板命令，那么在卸载模板之前，还要对所有已激活模板的节点使用以下接口解除模板：prefixPath

1. public void deactivateTemplateOn(String templateName, String prefixPath);

• 以上解除模板接口中，参数如果含有通配符（或）则按 PathPattern 匹配目标路径，否则仅表达其字面量对应的路径。prefixPath***
• 解除模板接口会删除对应节点按照模板中的序列写入的数据。

数据操作接口 DML

数据写入

推荐使用 insertTablet 帮助提高写入效率

• 插入一个 Tablet，Tablet 是一个设备若干行数据块，每一行的列都相同
  • 写入效率高
  • 支持写入空值：空值处可以填入任意值，然后通过 BitMap 标记空值

 复制代码

1. void insertTablet(Tablet tablet)
3. public class Tablet {
4. /** deviceId of this tablet */
5. public String prefixPath;
6. /** the list of measurement schemas for creating the tablet */
7. private List<MeasurementSchema> schemas;
8. /** timestamps in this tablet */
9. public long[] timestamps;
10. /** each object is a primitive type array, which represents values of one measurement */
11. public Object[] values;
12. /** each bitmap represents the existence of each value in the current column. */
13. public BitMap[] bitMaps;
14. /** the number of rows to include in this tablet */
15. public int rowSize;
16. /** the maximum number of rows for this tablet */
17. private int maxRowNumber;
18. /** whether this tablet store data of aligned timeseries or not */
19. private boolean isAligned;
20. }

• 插入多个 Tablet

1. void insertTablets(Map<String, Tablet> tablets)

• 插入一个 Record，一个 Record 是一个设备一个时间戳下多个测点的数据。这里的 value 是 Object 类型，相当于提供了一个公用接口，后面可以通过 TSDataType 将 value 强转为原类型

1. void insertRecord(String prefixPath, long time, List<String> measurements,
2. List<TSDataType> types, List<Object> values)

• 插入多个 Record

1. void insertRecords(List<String> deviceIds,
2. List<Long> times,
3. List<List<String>> measurementsList,
4. List<List<TSDataType>> typesList,
5. List<List<Object>> valuesList)

• 插入同属于一个 device 的多个 Record

1. void insertRecordsOfOneDevice(String deviceId, List<Long> times,
2. List<List<String>> measurementsList, List<List<TSDataType>> typesList,
3. List<List<Object>> valuesList)

带有类型推断的写入

当数据均是 String 类型时，我们可以使用如下接口，根据 value 的值进行类型推断。例如：value 为 “true” ，就可以自动推断为布尔类型。value 为 “3.2” ，就可以自动推断为数值类型。服务器需要做类型推断，可能会有额外耗时，速度较无需类型推断的写入慢

• 插入一个 Record，一个 Record 是一个设备一个时间戳下多个测点的数据

1. void insertRecord(String prefixPath, long time, List<String> measurements, List<String> values)

• 插入多个 Record

1. void insertRecords(List<String> deviceIds, List<Long> times,
2. List<List<String>> measurementsList, List<List<String>> valuesList)

• 插入同属于一个 device 的多个 Record

1. void insertStringRecordsOfOneDevice(String deviceId, List<Long> times,
2. List<List<String>> measurementsList, List<List<String>> valuesList)

对齐时间序列的写入

对齐时间序列的写入使用 insertAlignedXXX 接口，其余与上述接口类似：

• insertAlignedRecord
• insertAlignedRecords
• insertAlignedRecordsOfOneDevice
• insertAlignedStringRecordsOfOneDevice
• insertAlignedTablet
• insertAlignedTablets

数据删除

• 删除一个或多个时间序列在某个时间点前或这个时间点的数据

1. void deleteData(String path, long endTime)
2. void deleteData(List<String> paths, long endTime)

数据查询

• 原始数据查询。时间间隔包含开始时间，不包含结束时间

1. SessionDataSet executeRawDataQuery(List<String> paths, long startTime, long endTime)

• 查询最后一条时间戳大于等于某个时间点的数据

1. SessionDataSet executeLastDataQuery(List<String> paths, long LastTime)

IoTDB-SQL 接口

• 执行查询语句

1. SessionDataSet executeQueryStatement(String sql)

• 执行非查询语句

1. void executeNonQueryStatement(String sql)

写入测试接口 (用于分析网络带宽)

不实际写入数据，只将数据传输到 server 即返回

• 测试 insertRecord

1. void testInsertRecord(String deviceId, long time, List<String> measurements, List<String> values)
3. void testInsertRecord(String deviceId, long time, List<String> measurements,
4. List<TSDataType> types, List<Object> values)

• 测试 testInsertRecords

1. void testInsertRecords(List<String> deviceIds, List<Long> times,
2. List<List<String>> measurementsList, List<List<String>> valuesList)
4. void testInsertRecords(List<String> deviceIds, List<Long> times,
5. List<List<String>> measurementsList, List<List<TSDataType>> typesList,
6. List<List<Object>> valuesList)

• 测试 insertTablet

1. void testInsertTablet(Tablet tablet)

• 测试 insertTablets

1. void testInsertTablets(Map<String, Tablet> tablets)

示例代码

浏览上述接口的详细信息，请参阅代码 session/src/main/java/org/apache/iotdb/session/Session.java

使用上述接口的示例代码在 example/session/src/main/java/org/apache/iotdb/SessionExample.java

使用对齐时间序列和元数据模板的示例可以参见 example/session/src/main/java/org/apache/iotdb/AlignedTimeseriesSessionExample.java

针对原生接口的连接池

我们提供了一个针对原生接口的连接池 ()，使用该接口时，你只需要指定连接池的大小，就可以在使用时从池中获取连接。 如果超过 60s 都没得到一个连接的话，那么会打印一条警告日志，但是程序仍将继续等待。SessionPool

当一个连接被用完后，他会自动返回池中等待下次被使用； 当一个连接损坏后，他会从池中被删除，并重建一个连接重新执行用户的操作； 你还可以像创建 Session 那样在创建 SessionPool 时指定多个可连接节点的 url，以保证分布式集群中客户端的高可用性。

对于查询操作：

1. 使用 SessionPool 进行查询时，得到的结果集是的封装类SessionDataSetSessionDataSetWrapper;
2. 若对于一个查询的结果集，用户并没有遍历完且不再想继续遍历时，需要手动调用释放连接的操作closeResultSet;
3. 若对一个查询的结果集遍历时出现异常，也需要手动调用释放连接的操作.closeResultSet
4. 可以调用 的 方法得到结果集列名SessionDataSetWrappergetColumnNames()

使用示例可以参见 session/src/test/java/org/apache/iotdb/session/pool/SessionPoolTest.java

或 example/session/src/main/java/org/apache/iotdb/SessionPoolExample.java

集群信息相关的接口 （仅在集群模式下可用）

集群信息相关的接口允许用户获取如数据分区情况、节点是否当机等信息。 要使用该 API，需要增加依赖：

1. <dependencies>
2. <dependency>
3. <groupId>org.apache.iotdb</groupId>
4. <artifactId>iotdb-thrift-cluster</artifactId>
5. <version>0.13.0-SNAPSHOT</version>
6. </dependency>
7. </dependencies>

建立连接与关闭连接的示例：

1. import org.apache.thrift.protocol.TBinaryProtocol;
2. import org.apache.thrift.transport.TSocket;
3. import org.apache.thrift.transport.TTransport;
4. import org.apache.thrift.transport.TTransportException;
5. import org.apache.iotdb.rpc.RpcTransportFactory;
7. public class CluserInfoClient {
8. TTransport transport;
9. ClusterInfoService.Client client;
10. public void connect() {
11. transport =
12. RpcTransportFactory.INSTANCE.getTransport(
13. new TSocket(
14. // the RPC address
15. IoTDBDescriptor.getInstance().getConfig().getRpcAddress(),
16. // the RPC port
17. ClusterDescriptor.getInstance().getConfig().getClusterRpcPort()));
18. try {
19. transport.open();
20. } catch (TTransportException e) {
21. Assert.fail(e.getMessage());
22. }
23. //get the client
24. client = new ClusterInfoService.Client(new TBinaryProtocol(transport));
25. }
26. public void close() {
27. transport.close();
28. }
29. }

API 列表：

• 获取集群中的各个节点的信息（构成哈希环）

1. list<Node> getRing();

• 给定一个路径（应包括一个 SG 作为前缀）和起止时间，获取其覆盖的数据分区情况：

1. /**
2. * @param path input path (should contains a Storage group name as its prefix)
3. * @return the data partition info. If the time range only covers one data partition, the the size
4. * of the list is one.
5. */
6. list<DataPartitionEntry> getDataPartition(1:string path, 2:long startTime, 3:long endTime);

• 给定一个路径（应包括一个 SG 作为前缀），获取其被分到了哪个节点上：

1. /**
2. * @param path input path (should contains a Storage group name as its prefix)
3. * @return metadata partition information
4. */
5. list<Node> getMetaPartition(1:string path);

• 获取所有节点的死活状态：

1. /**
2. * @return key: node, value: live or not
3. */
4. map<Node, bool> getAllNodeStatus();

• 获取当前连接节点的 Raft 组信息（投票编号等）（一般用户无需使用该接口）:

1. /**
2. * @return A multi-line string with each line representing the total time consumption, invocation
3. * number, and average time consumption.
4. */
5. string getInstrumentingInfo();