Skip to content

Commit 76b818d

Browse files
CritasWangCritasWang
committed
Bump tsfile version to 2.1.1
1 parent bf06d2e commit 76b818d

File tree

21 files changed

+702
-618
lines changed

21 files changed

+702
-618
lines changed

‎connectors/flink-iotdb-connector/src/main/java/org/apache/iotdb/flink/IoTDBSink.java‎

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -78,14 +78,13 @@ public void open(Configuration parameters) throws Exception {
7878

7979
void initSession() {
8080
if (options.getNodeUrls() != null) {
81-
pool = new SessionPool(
82-
options.getNodeUrls(),
83-
options.getUser(),
84-
options.getPassword(),
85-
sessionPoolSize);
81+
pool =
82+
new SessionPool(
83+
options.getNodeUrls(), options.getUser(), options.getPassword(), sessionPoolSize);
8684
return;
8785
}
88-
pool = new SessionPool(
86+
pool =
87+
new SessionPool(
8988
options.getHost(),
9089
options.getPort(),
9190
options.getUser(),

‎connectors/flink-iotdb-connector/src/main/java/org/apache/iotdb/flink/options/IoTDBSinkOptions.java‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -43,10 +43,10 @@ public IoTDBSinkOptions(
4343
}
4444

4545
public IoTDBSinkOptions(
46-
List<String> nodeUrls,
47-
String user,
48-
String password,
49-
List<TimeseriesOption> timeseriesOptionList) {
46+
List<String> nodeUrls,
47+
String user,
48+
String password,
49+
List<TimeseriesOption> timeseriesOptionList) {
5050
super(nodeUrls, user, password);
5151
this.timeseriesOptionList = timeseriesOptionList;
5252
}

‎connectors/spark-iotdb-table-connector/spark-iotdb-table-common/src/main/scala/org/apache/iotdb/spark/table/db/IoTDBUtils.scala‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ import org.apache.spark.unsafe.types.UTF8String
2828
import org.apache.tsfile.enums.TSDataType
2929
import org.apache.tsfile.read.common.RowRecord
3030
import org.apache.tsfile.utils.{Binary, DateUtils}
31-
import org.apache.tsfile.write.record.Tablet.ColumnCategory
31+
import org.apache.tsfile.enums.ColumnCategory
3232

3333
import java.util
3434

‎connectors/spark-iotdb-table-connector/spark-iotdb-table-common/src/main/scala/org/apache/iotdb/spark/table/db/write/IoTDBDataWriter.scala‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ import org.apache.spark.sql.connector.write.{DataWriter, WriterCommitMessage}
2828
import org.apache.spark.sql.types.{NullType, StructType}
2929
import org.apache.tsfile.enums.TSDataType
3030
import org.apache.tsfile.write.record.Tablet
31-
import org.apache.tsfile.write.record.Tablet.ColumnCategory
31+
import org.apache.tsfile.enums.ColumnCategory
3232

3333
class IoTDBDataWriter(options: IoTDBOptions, writeSchema: StructType, tableSchema: StructType) extends DataWriter[InternalRow] with Logging {
3434

‎examples/kafka/src/main/java/org/apache/iotdb/kafka/relational/RelationalConstant.java‎

Lines changed: 35 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -21,29 +21,39 @@
2121

2222
public class RelationalConstant {
2323

24-
public static final String KAFKA_SERVICE_URL = "172.20.31.71:9094";
25-
public static final String TOPIC = "Kafka-Relational-Test";
26-
public static final String[] IOTDB_URLS = {
27-
"127.0.0.1:6667"
28-
};
29-
public static final String IOTDB_USERNAME = "root";
30-
public static final String IOTDB_PASSWORD = "root";
31-
public static final int SESSION_SIZE = 3;
32-
public static final int CONSUMER_THREAD_NUM = 5;
33-
public static final String[] DATABASES = {"kafka_db1", "kafka_db2"};
34-
public static final String[][] TABLES = {
35-
// database, tableName, columnNames, columnTypes, columnCategories
36-
{"kafka_db1", "tb1", "time,region,model_id,temperature,status", "TIMESTAMP,STRING,STRING,FLOAT,BOOLEAN", "TIME,TAG,ATTRIBUTE,FIELD,FIELD"},
37-
{"kafka_db2", "tb2", "time,plant_id,humidity,status", "TIMESTAMP,STRING,FLOAT,BOOLEAN", "TIME,TAG,FIELD,FIELD"}
38-
};
39-
public static final String[] ALL_DATA = {
40-
// database;tableName;columnName[,columnName]*;value[,value]*[,value[:value]*]*
41-
"kafka_db1;tb1;time,temperature,status;17,3.26,true;18,3.27,false;19,3.28,true",
42-
"kafka_db1;tb1;time,region,model_id,temperature;20,'rgn1','id1',3.31",
43-
"kafka_db2;tb2;time,plant_id,humidity,status;50,'id1',68.7,true",
44-
"kafka_db2;tb2;time,plant_id,humidity,status;51,'id2',68.5,false",
45-
"kafka_db2;tb2;time,plant_id,humidity,status;52,'id3',68.3,true",
46-
"kafka_db2;tb2;time,plant_id,humidity,status;53,'id4',68.8,true",
47-
"kafka_db2;tb2;time,plant_id,humidity,status;54,'id5',68.9,true"
48-
};
24+
public static final String KAFKA_SERVICE_URL = "172.20.31.71:9094";
25+
public static final String TOPIC = "Kafka-Relational-Test";
26+
public static final String[] IOTDB_URLS = {"127.0.0.1:6667"};
27+
public static final String IOTDB_USERNAME = "root";
28+
public static final String IOTDB_PASSWORD = "root";
29+
public static final int SESSION_SIZE = 3;
30+
public static final int CONSUMER_THREAD_NUM = 5;
31+
public static final String[] DATABASES = {"kafka_db1", "kafka_db2"};
32+
public static final String[][] TABLES = {
33+
// database, tableName, columnNames, columnTypes, columnCategories
34+
{
35+
"kafka_db1",
36+
"tb1",
37+
"time,region,model_id,temperature,status",
38+
"TIMESTAMP,STRING,STRING,FLOAT,BOOLEAN",
39+
"TIME,TAG,ATTRIBUTE,FIELD,FIELD"
40+
},
41+
{
42+
"kafka_db2",
43+
"tb2",
44+
"time,plant_id,humidity,status",
45+
"TIMESTAMP,STRING,FLOAT,BOOLEAN",
46+
"TIME,TAG,FIELD,FIELD"
47+
}
48+
};
49+
public static final String[] ALL_DATA = {
50+
// database;tableName;columnName[,columnName]*;value[,value]*[,value[:value]*]*
51+
"kafka_db1;tb1;time,temperature,status;17,3.26,true;18,3.27,false;19,3.28,true",
52+
"kafka_db1;tb1;time,region,model_id,temperature;20,'rgn1','id1',3.31",
53+
"kafka_db2;tb2;time,plant_id,humidity,status;50,'id1',68.7,true",
54+
"kafka_db2;tb2;time,plant_id,humidity,status;51,'id2',68.5,false",
55+
"kafka_db2;tb2;time,plant_id,humidity,status;52,'id3',68.3,true",
56+
"kafka_db2;tb2;time,plant_id,humidity,status;53,'id4',68.8,true",
57+
"kafka_db2;tb2;time,plant_id,humidity,status;54,'id5',68.9,true"
58+
};
4959
}

‎examples/kafka/src/main/java/org/apache/iotdb/kafka/relational/RelationalConsumer.java‎

Lines changed: 114 additions & 104 deletions
Original file line numberDiff line numberDiff line change
@@ -24,126 +24,136 @@
2424
import org.apache.iotdb.rpc.IoTDBConnectionException;
2525
import org.apache.iotdb.rpc.StatementExecutionException;
2626
import org.apache.iotdb.session.pool.TableSessionPoolBuilder;
27+
2728
import org.apache.kafka.clients.consumer.ConsumerConfig;
2829
import org.apache.kafka.clients.consumer.KafkaConsumer;
2930
import org.apache.kafka.common.serialization.StringDeserializer;
3031
import org.slf4j.Logger;
3132
import org.slf4j.LoggerFactory;
3233

33-
import java.util.*;
34+
import java.util.ArrayList;
35+
import java.util.Arrays;
36+
import java.util.Collections;
37+
import java.util.List;
38+
import java.util.Properties;
3439
import java.util.concurrent.ExecutorService;
3540
import java.util.concurrent.Executors;
3641

3742
public class RelationalConsumer {
3843

39-
private static final Logger LOGGER = LoggerFactory.getLogger(RelationalConsumer.class);
40-
private static ITableSessionPool tableSessionPool;
41-
private List<KafkaConsumer<String, String>> consumerList;
42-
43-
private RelationalConsumer(List<KafkaConsumer<String, String>> consumerList) {
44-
this.consumerList = consumerList;
45-
initSessionPool();
44+
private static final Logger LOGGER = LoggerFactory.getLogger(RelationalConsumer.class);
45+
private static ITableSessionPool tableSessionPool;
46+
private List<KafkaConsumer<String, String>> consumerList;
47+
48+
private RelationalConsumer(List<KafkaConsumer<String, String>> consumerList) {
49+
this.consumerList = consumerList;
50+
initSessionPool();
51+
}
52+
53+
private static void initSessionPool() {
54+
tableSessionPool =
55+
new TableSessionPoolBuilder()
56+
.nodeUrls(Arrays.asList(RelationalConstant.IOTDB_URLS))
57+
.user(RelationalConstant.IOTDB_USERNAME)
58+
.password(RelationalConstant.IOTDB_PASSWORD)
59+
.maxSize(RelationalConstant.SESSION_SIZE)
60+
.build();
61+
}
62+
63+
public static void main(String[] args) {
64+
List<KafkaConsumer<String, String>> consumerList = new ArrayList<>();
65+
for (int i = 0; i < RelationalConstant.CONSUMER_THREAD_NUM; i++) {
66+
Properties props = new Properties();
67+
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, RelationalConstant.KAFKA_SERVICE_URL);
68+
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
69+
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
70+
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
71+
props.put(ConsumerConfig.GROUP_ID_CONFIG, RelationalConstant.TOPIC);
72+
73+
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
74+
consumerList.add(consumer);
75+
consumer.subscribe(Collections.singleton(RelationalConstant.TOPIC));
4676
}
47-
48-
private static void initSessionPool() {
49-
tableSessionPool =
50-
new TableSessionPoolBuilder()
51-
.nodeUrls(Arrays.asList(RelationalConstant.IOTDB_URLS))
52-
.user(RelationalConstant.IOTDB_USERNAME)
53-
.password(RelationalConstant.IOTDB_PASSWORD)
54-
.maxSize(RelationalConstant.SESSION_SIZE)
55-
.build();
56-
}
57-
58-
public static void main(String[] args) {
59-
List<KafkaConsumer<String, String>> consumerList = new ArrayList<>();
60-
for (int i = 0; i < RelationalConstant.CONSUMER_THREAD_NUM; i++) {
61-
Properties props = new Properties();
62-
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, RelationalConstant.KAFKA_SERVICE_URL);
63-
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
64-
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
65-
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
66-
props.put(ConsumerConfig.GROUP_ID_CONFIG, RelationalConstant.TOPIC);
67-
68-
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
69-
consumerList.add(consumer);
70-
consumer.subscribe(Collections.singleton(RelationalConstant.TOPIC));
71-
}
72-
RelationalConsumer consumer = new RelationalConsumer(consumerList);
73-
initIoTDB();
74-
consumer.consumeInParallel();
77+
RelationalConsumer consumer = new RelationalConsumer(consumerList);
78+
initIoTDB();
79+
consumer.consumeInParallel();
80+
}
81+
82+
private static void initIoTDB() {
83+
for (String db : RelationalConstant.DATABASES) {
84+
boolean res = createDatabase(db);
85+
if (!res) {
86+
throw new RuntimeException("Create database failed");
87+
}
7588
}
76-
77-
private static void initIoTDB() {
78-
for (String db : RelationalConstant.DATABASES) {
79-
boolean res = createDatabase(db);
80-
if (!res) {
81-
throw new RuntimeException("Create database failed");
82-
}
83-
}
84-
for (String[] tableInfo : RelationalConstant.TABLES) {
85-
boolean res = createTable(tableInfo);
86-
if (!res) {
87-
throw new RuntimeException("Create table failed");
88-
}
89-
}
89+
for (String[] tableInfo : RelationalConstant.TABLES) {
90+
boolean res = createTable(tableInfo);
91+
if (!res) {
92+
throw new RuntimeException("Create table failed");
93+
}
9094
}
91-
92-
private static boolean createDatabase(String dbName) {
93-
try (ITableSession session = tableSessionPool.getSession()) {
94-
try {
95-
session.executeNonQueryStatement(String.format("CREATE DATABASE %s", dbName));
96-
} catch (IoTDBConnectionException | StatementExecutionException e) {
97-
LOGGER.error("Create Database Error: ", e);
98-
return false;
99-
}
100-
} catch (IoTDBConnectionException e) {
101-
LOGGER.error("Get Table Session Error: ", e);
102-
return false;
103-
}
104-
return true;
95+
}
96+
97+
private static boolean createDatabase(String dbName) {
98+
try (ITableSession session = tableSessionPool.getSession()) {
99+
try {
100+
session.executeNonQueryStatement(String.format("CREATE DATABASE %s", dbName));
101+
} catch (IoTDBConnectionException | StatementExecutionException e) {
102+
LOGGER.error("Create Database Error: ", e);
103+
return false;
104+
}
105+
} catch (IoTDBConnectionException e) {
106+
LOGGER.error("Get Table Session Error: ", e);
107+
return false;
105108
}
106-
107-
private static boolean createTable(String[] tableInfo) {
108-
try (ITableSession session = tableSessionPool.getSession()) {
109-
String sql = getCreateTableSQL(tableInfo);
110-
try {
111-
session.executeNonQueryStatement(sql);
112-
} catch (IoTDBConnectionException | StatementExecutionException e) {
113-
LOGGER.error("Create Table Error: ", e);
114-
return false;
115-
}
116-
} catch (IoTDBConnectionException e) {
117-
LOGGER.error("Get Table Session Error: ", e);
118-
return false;
119-
}
120-
return true;
109+
return true;
110+
}
111+
112+
private static boolean createTable(String[] tableInfo) {
113+
try (ITableSession session = tableSessionPool.getSession()) {
114+
String sql = getCreateTableSQL(tableInfo);
115+
try {
116+
session.executeNonQueryStatement(sql);
117+
} catch (IoTDBConnectionException | StatementExecutionException e) {
118+
LOGGER.error("Create Table Error: ", e);
119+
return false;
120+
}
121+
} catch (IoTDBConnectionException e) {
122+
LOGGER.error("Get Table Session Error: ", e);
123+
return false;
121124
}
122-
123-
private static String getCreateTableSQL(String[] tableInfo) {
124-
StringBuilder sql = new StringBuilder();
125-
sql.append("CREATE TABLE \"").append(tableInfo[0]).append("\".\"").append(tableInfo[1]).append("\" (");
126-
127-
String[] columnNames = tableInfo[2].split(",");
128-
String[] columnTypes = tableInfo[3].split(",");
129-
String[] columnCategories = tableInfo[4].split(",");
130-
int columnSize = columnNames.length;
131-
132-
for (int i = 0; i < columnSize; i++) {
133-
sql.append(columnNames[i]).append(" ");
134-
sql.append(columnTypes[i]).append(" ");
135-
sql.append(columnCategories[i]).append(",");
136-
}
137-
sql.deleteCharAt(sql.length() - 1);
138-
sql.append(")");
139-
return sql.toString();
125+
return true;
126+
}
127+
128+
private static String getCreateTableSQL(String[] tableInfo) {
129+
StringBuilder sql = new StringBuilder();
130+
sql.append("CREATE TABLE \"")
131+
.append(tableInfo[0])
132+
.append("\".\"")
133+
.append(tableInfo[1])
134+
.append("\" (");
135+
136+
String[] columnNames = tableInfo[2].split(",");
137+
String[] columnTypes = tableInfo[3].split(",");
138+
String[] columnCategories = tableInfo[4].split(",");
139+
int columnSize = columnNames.length;
140+
141+
for (int i = 0; i < columnSize; i++) {
142+
sql.append(columnNames[i]).append(" ");
143+
sql.append(columnTypes[i]).append(" ");
144+
sql.append(columnCategories[i]).append(",");
140145
}
141-
142-
private void consumeInParallel() {
143-
ExecutorService executor = Executors.newFixedThreadPool(RelationalConstant.CONSUMER_THREAD_NUM);
144-
for (int i = 0; i < consumerList.size(); i++) {
145-
RelationalConsumerThread consumerThread = new RelationalConsumerThread(consumerList.get(i), tableSessionPool);
146-
executor.submit(consumerThread);
147-
}
146+
sql.deleteCharAt(sql.length() - 1);
147+
sql.append(")");
148+
return sql.toString();
149+
}
150+
151+
private void consumeInParallel() {
152+
ExecutorService executor = Executors.newFixedThreadPool(RelationalConstant.CONSUMER_THREAD_NUM);
153+
for (int i = 0; i < consumerList.size(); i++) {
154+
RelationalConsumerThread consumerThread =
155+
new RelationalConsumerThread(consumerList.get(i), tableSessionPool);
156+
executor.submit(consumerThread);
148157
}
158+
}
149159
}

0 commit comments

Comments
 (0)