You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by qi...@apache.org on 2020/05/06 02:22:46 UTC
[incubator-iotdb] 01/04: [IOTDB-560] add TSRecordOutputFormat to
write TsFile via Flink DataSet/DataStream API.
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 99975f716f97fefe01831bcc4b3e02812dfc0bcf
Author: Wei Zhong <we...@gmail.com>
AuthorDate: Tue Apr 21 21:07:14 2020 +0800
[IOTDB-560] add TSRecordOutputFormat to write TsFile via Flink DataSet/DataStream API.
---
example/flink/README.md | 2 +
.../apache/iotdb/flink/FlinkTsFileBatchSink.java | 114 +++++++++++++++
.../apache/iotdb/flink/FlinkTsFileStreamSink.java | 115 +++++++++++++++
.../java/org/apache/iotdb/flink/TsFlieUtils.java | 26 ++++
flink-tsfile-connector/README.md | 86 ++++++++++-
.../iotdb/flink/tsfile/RowTSRecordConverter.java | 157 +++++++++++++++++++++
.../iotdb/flink/tsfile/TSRecordConverter.java | 36 +++++
.../iotdb/flink/tsfile/TSRecordOutputFormat.java | 99 +++++++++++++
.../iotdb/flink/tsfile/TsFileOutputFormat.java | 118 ++++++++++++++++
.../tsfile/RowTSRecordOutputFormatITCase.java | 56 ++++++++
.../flink/tsfile/RowTSRecordOutputFormatTest.java | 77 ++++++++++
.../flink/tsfile/RowTsFileConnectorTestBase.java | 87 ++++++++++++
.../flink/tsfile/RowTsFileInputFormatTestBase.java | 72 +---------
.../tsfile/RowTsFileOutputFormatTestBase.java | 117 +++++++++++++++
.../apache/iotdb/flink/util/TsFileWriteUtil.java | 2 +-
.../apache/iotdb/tsfile/write/schema/Schema.java | 4 +-
16 files changed, 1097 insertions(+), 71 deletions(-)
diff --git a/example/flink/README.md b/example/flink/README.md
index 5059946..38a6d33 100644
--- a/example/flink/README.md
+++ b/example/flink/README.md
@@ -36,3 +36,5 @@ The example is to show how to send data to a IoTDB server from a Flink job.
* Run `org.apache.iotdb.flink.FlinkTsFileBatchSource.java` to create a tsfile and read it via a flink DataSet job on local mini cluster.
* Run `org.apache.iotdb.flink.FlinkTsFileStreamSource.java` to create a tsfile and read it via a flink DataStream job on local mini cluster.
+* Run `org.apache.iotdb.flink.FlinkTsFileBatchSink.java` to write a tsfile via a flink DataSet job on local mini cluster and print the content to stdout.
+* Run `org.apache.iotdb.flink.FlinkTsFileStreamSink.java` to write a tsfile via a flink DataStream job on local mini cluster and print the content to stdout.
diff --git a/example/flink/src/main/java/org/apache/iotdb/flink/FlinkTsFileBatchSink.java b/example/flink/src/main/java/org/apache/iotdb/flink/FlinkTsFileBatchSink.java
new file mode 100644
index 0000000..28aaf83
--- /dev/null
+++ b/example/flink/src/main/java/org/apache/iotdb/flink/FlinkTsFileBatchSink.java
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.iotdb.flink;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.tuple.Tuple7;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.types.Row;
+import org.apache.iotdb.flink.tsfile.RowTSRecordConverter;
+import org.apache.iotdb.flink.tsfile.TSRecordOutputFormat;
+import org.apache.iotdb.tsfile.common.constant.QueryConstant;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.iotdb.tsfile.write.schema.Schema;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * The example of writing TsFile via Flink DataSet API.
+ */
+public class FlinkTsFileBatchSink {
+
+ public static final String DEFAULT_TEMPLATE = "template";
+
+ public static void main(String[] arg) throws Exception {
+ String path = new File("test.tsfile").getAbsolutePath();
+ new File(path).deleteOnExit();
+ String[] filedNames = {
+ QueryConstant.RESERVED_TIME,
+ "device_1.sensor_1",
+ "device_1.sensor_2",
+ "device_1.sensor_3",
+ "device_2.sensor_1",
+ "device_2.sensor_2",
+ "device_2.sensor_3"
+ };
+ TypeInformation[] typeInformations = new TypeInformation[] {
+ Types.LONG,
+ Types.LONG,
+ Types.LONG,
+ Types.LONG,
+ Types.LONG,
+ Types.LONG,
+ Types.LONG
+ };
+ RowTypeInfo rowTypeInfo = new RowTypeInfo(typeInformations, filedNames);
+ Schema schema = new Schema();
+ schema.extendTemplate(
+ DEFAULT_TEMPLATE, new MeasurementSchema("sensor_1", TSDataType.INT64, TSEncoding.TS_2DIFF));
+ schema.extendTemplate(
+ DEFAULT_TEMPLATE, new MeasurementSchema("sensor_2", TSDataType.INT64, TSEncoding.TS_2DIFF));
+ schema.extendTemplate(
+ DEFAULT_TEMPLATE, new MeasurementSchema("sensor_3", TSDataType.INT64, TSEncoding.TS_2DIFF));
+ RowTSRecordConverter converter = new RowTSRecordConverter(rowTypeInfo);
+ TSRecordOutputFormat<Row> outputFormat = new TSRecordOutputFormat<>(schema, converter);
+
+ List<Tuple7> data = new ArrayList<>(7);
+ data.add(new Tuple7(1L, 2L, 3L, 4L, 5L, 6L, 7L));
+ data.add(new Tuple7(2L, 3L, 4L, 5L, 6L, 7L, 8L));
+ data.add(new Tuple7(3L, 4L, 5L, 6L, 7L, 8L, 9L));
+ data.add(new Tuple7(4L, 5L, 6L, 7L, 8L, 9L, 10L));
+ data.add(new Tuple7(6L, 6L, 7L, 8L, 9L, 10L, 11L));
+ data.add(new Tuple7(7L, 7L, 8L, 9L, 10L, 11L, 12L));
+ data.add(new Tuple7(8L, 8L, 9L, 10L, 11L, 12L, 13L));
+
+ ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ // If the parallelism > 1, flink create a directory and each subtask will create a tsfile under the directory.
+ env.setParallelism(1);
+ DataSet<Tuple7> source = env.fromCollection(
+ data, Types.TUPLE(Types.LONG, Types.LONG, Types.LONG, Types.LONG, Types.LONG, Types.LONG, Types.LONG));
+ source.map(t -> {
+ Row row = new Row(7);
+ for (int i = 0; i < 7; i++) {
+ row.setField(i, t.getField(i));
+ }
+ return row;
+ }).returns(rowTypeInfo).write(outputFormat, path);
+
+ env.execute();
+
+ List<Path> paths = Arrays.stream(filedNames)
+ .filter(s -> !s.equals(QueryConstant.RESERVED_TIME))
+ .map(Path::new)
+ .collect(Collectors.toList());
+ String[] result = TsFlieUtils.readTsFile(path, paths);
+ for (String row : result) {
+ System.out.println(row);
+ }
+ }
+}
diff --git a/example/flink/src/main/java/org/apache/iotdb/flink/FlinkTsFileStreamSink.java b/example/flink/src/main/java/org/apache/iotdb/flink/FlinkTsFileStreamSink.java
new file mode 100644
index 0000000..2043c87
--- /dev/null
+++ b/example/flink/src/main/java/org/apache/iotdb/flink/FlinkTsFileStreamSink.java
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.iotdb.flink;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.tuple.Tuple7;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.types.Row;
+import org.apache.iotdb.flink.tsfile.RowTSRecordConverter;
+import org.apache.iotdb.flink.tsfile.TSRecordOutputFormat;
+import org.apache.iotdb.tsfile.common.constant.QueryConstant;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.iotdb.tsfile.write.schema.Schema;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * The example of writing TsFile via Flink DataStream API.
+ */
+public class FlinkTsFileStreamSink {
+
+ public static final String DEFAULT_TEMPLATE = "template";
+
+ public static void main(String[] arg) throws Exception {
+ String path = new File("test.tsfile").getAbsolutePath();
+ new File(path).deleteOnExit();
+ String[] filedNames = {
+ QueryConstant.RESERVED_TIME,
+ "device_1.sensor_1",
+ "device_1.sensor_2",
+ "device_1.sensor_3",
+ "device_2.sensor_1",
+ "device_2.sensor_2",
+ "device_2.sensor_3"
+ };
+ TypeInformation[] typeInformations = new TypeInformation[] {
+ Types.LONG,
+ Types.LONG,
+ Types.LONG,
+ Types.LONG,
+ Types.LONG,
+ Types.LONG,
+ Types.LONG
+ };
+ RowTypeInfo rowTypeInfo = new RowTypeInfo(typeInformations, filedNames);
+ Schema schema = new Schema();
+ schema.extendTemplate(
+ DEFAULT_TEMPLATE, new MeasurementSchema("sensor_1", TSDataType.INT64, TSEncoding.TS_2DIFF));
+ schema.extendTemplate(
+ DEFAULT_TEMPLATE, new MeasurementSchema("sensor_2", TSDataType.INT64, TSEncoding.TS_2DIFF));
+ schema.extendTemplate(
+ DEFAULT_TEMPLATE, new MeasurementSchema("sensor_3", TSDataType.INT64, TSEncoding.TS_2DIFF));
+ RowTSRecordConverter converter = new RowTSRecordConverter(rowTypeInfo);
+ TSRecordOutputFormat<Row> outputFormat = new TSRecordOutputFormat<>(schema, converter);
+ outputFormat.setOutputFilePath(new org.apache.flink.core.fs.Path(path));
+
+ List<Tuple7> data = new ArrayList<>(7);
+ data.add(new Tuple7(1L, 2L, 3L, 4L, 5L, 6L, 7L));
+ data.add(new Tuple7(2L, 3L, 4L, 5L, 6L, 7L, 8L));
+ data.add(new Tuple7(3L, 4L, 5L, 6L, 7L, 8L, 9L));
+ data.add(new Tuple7(4L, 5L, 6L, 7L, 8L, 9L, 10L));
+ data.add(new Tuple7(6L, 6L, 7L, 8L, 9L, 10L, 11L));
+ data.add(new Tuple7(7L, 7L, 8L, 9L, 10L, 11L, 12L));
+ data.add(new Tuple7(8L, 8L, 9L, 10L, 11L, 12L, 13L));
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ // If the parallelism > 1, flink create a directory and each subtask will create a tsfile under the directory.
+ env.setParallelism(1);
+ DataStream<Tuple7> source = env.fromCollection(
+ data, Types.TUPLE(Types.LONG, Types.LONG, Types.LONG, Types.LONG, Types.LONG, Types.LONG, Types.LONG));
+ source.map(t -> {
+ Row row = new Row(7);
+ for (int i = 0; i < 7; i++) {
+ row.setField(i, t.getField(i));
+ }
+ return row;
+ }).returns(rowTypeInfo).writeUsingOutputFormat(outputFormat);
+
+ env.execute();
+
+ List<Path> paths = Arrays.stream(filedNames)
+ .filter(s -> !s.equals(QueryConstant.RESERVED_TIME))
+ .map(Path::new)
+ .collect(Collectors.toList());
+ String[] result = TsFlieUtils.readTsFile(path, paths);
+ for (String row : result) {
+ System.out.println(row);
+ }
+ }
+}
diff --git a/example/flink/src/main/java/org/apache/iotdb/flink/TsFlieUtils.java b/example/flink/src/main/java/org/apache/iotdb/flink/TsFlieUtils.java
index 5b1ffba..f97b683 100644
--- a/example/flink/src/main/java/org/apache/iotdb/flink/TsFlieUtils.java
+++ b/example/flink/src/main/java/org/apache/iotdb/flink/TsFlieUtils.java
@@ -20,6 +20,12 @@ package org.apache.iotdb.flink;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.iotdb.tsfile.read.ReadOnlyTsFile;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.expression.QueryExpression;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
import org.apache.iotdb.tsfile.write.TsFileWriter;
import org.apache.iotdb.tsfile.write.record.TSRecord;
import org.apache.iotdb.tsfile.write.record.datapoint.DataPoint;
@@ -28,6 +34,10 @@ import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.apache.iotdb.tsfile.write.schema.Schema;
import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
/**
* Utils used to prepare source TsFiles for the examples.
@@ -69,4 +79,20 @@ public class TsFlieUtils {
System.out.println(e.getMessage());
}
}
+
+ public static String[] readTsFile(String tsFilePath, List<Path> paths) throws IOException {
+ QueryExpression expression = QueryExpression.create(paths, null);
+ TsFileSequenceReader reader = new TsFileSequenceReader(tsFilePath);
+ ReadOnlyTsFile readTsFile = new ReadOnlyTsFile(reader);
+ QueryDataSet queryDataSet = readTsFile.query(expression);
+ List<String> result = new ArrayList<>();
+ while (queryDataSet.hasNext()) {
+ RowRecord rowRecord = queryDataSet.next();
+ String row = rowRecord.getFields().stream()
+ .map(f -> f == null ? "null" : f.getStringValue())
+ .collect(Collectors.joining(","));
+ result.add(rowRecord.getTimestamp() + "," + row);
+ }
+ return result.toArray(new String[0]);
+ }
}
diff --git a/flink-tsfile-connector/README.md b/flink-tsfile-connector/README.md
index a571de7..8b8b88d 100644
--- a/flink-tsfile-connector/README.md
+++ b/flink-tsfile-connector/README.md
@@ -23,7 +23,7 @@
## 1. About TsFile-Flink-Connector
TsFile-Flink-Connector implements the support of Flink for external data sources of Tsfile type.
-This enables users to read Tsfile by Flink via DataStream/DataSet API.
+This enables users to read and write Tsfile by Flink via DataStream/DataSet API.
With this connector, you can
* load a single TsFile or multiple TsFiles(only for DataSet), from either the local file system or hdfs, into Flink
@@ -91,3 +91,87 @@ for (String s : result) {
}
```
+### TSRecordOutputFormat Example
+
+1. create TSRecordOutputFormat with default RowTSRecordConverter.
+
+```java
+String[] filedNames = {
+ QueryConstant.RESERVED_TIME,
+ "device_1.sensor_1",
+ "device_1.sensor_2",
+ "device_1.sensor_3",
+ "device_2.sensor_1",
+ "device_2.sensor_2",
+ "device_2.sensor_3"
+};
+TypeInformation[] typeInformations = new TypeInformation[] {
+ Types.LONG,
+ Types.FLOAT,
+ Types.INT,
+ Types.INT,
+ Types.FLOAT,
+ Types.INT,
+ Types.INT
+};
+RowTypeInfo rowTypeInfo = new RowTypeInfo(typeInformations, filedNames);
+Schema schema = new Schema();
+schema.extendTemplate("template", new MeasurementSchema("sensor_1", TSDataType.INT64, TSEncoding.TS_2DIFF));
+schema.extendTemplate("template", new MeasurementSchema("sensor_2", TSDataType.INT64, TSEncoding.TS_2DIFF));
+schema.extendTemplate("template", new MeasurementSchema("sensor_3", TSDataType.INT64, TSEncoding.TS_2DIFF));
+RowTSRecordConverter converter = new RowTSRecordConverter(rowTypeInfo);
+TSRecordOutputFormat<Row> outputFormat = new TSRecordOutputFormat<>(schema, converter);
+```
+
+2. write data via the output format:
+
+DataStream:
+
+```java
+StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
+senv.setParallelism(1);
+List<Tuple7> data = new ArrayList<>(7);
+data.add(new Tuple7(1L, 2L, 3L, 4L, 5L, 6L, 7L));
+data.add(new Tuple7(2L, 3L, 4L, 5L, 6L, 7L, 8L));
+data.add(new Tuple7(3L, 4L, 5L, 6L, 7L, 8L, 9L));
+data.add(new Tuple7(4L, 5L, 6L, 7L, 8L, 9L, 10L));
+data.add(new Tuple7(6L, 6L, 7L, 8L, 9L, 10L, 11L));
+data.add(new Tuple7(7L, 7L, 8L, 9L, 10L, 11L, 12L));
+data.add(new Tuple7(8L, 8L, 9L, 10L, 11L, 12L, 13L));
+outputFormat.setOutputFilePath(new org.apache.flink.core.fs.Path(path));
+DataStream<Tuple7> source = senv.fromCollection(
+ data, Types.TUPLE(Types.LONG, Types.LONG, Types.LONG, Types.LONG, Types.LONG, Types.LONG, Types.LONG));
+source.map(t -> {
+ Row row = new Row(7);
+ for (int i = 0; i < 7; i++) {
+ row.setField(i, t.getField(i));
+ }
+ return row;
+}).returns(rowTypeInfo).writeUsingOutputFormat(outputFormat);
+senv.execute();
+```
+
+DataSet:
+
+```java
+ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+env.setParallelism(1);
+List<Tuple7> data = new ArrayList<>(7);
+data.add(new Tuple7(1L, 2L, 3L, 4L, 5L, 6L, 7L));
+data.add(new Tuple7(2L, 3L, 4L, 5L, 6L, 7L, 8L));
+data.add(new Tuple7(3L, 4L, 5L, 6L, 7L, 8L, 9L));
+data.add(new Tuple7(4L, 5L, 6L, 7L, 8L, 9L, 10L));
+data.add(new Tuple7(6L, 6L, 7L, 8L, 9L, 10L, 11L));
+data.add(new Tuple7(7L, 7L, 8L, 9L, 10L, 11L, 12L));
+data.add(new Tuple7(8L, 8L, 9L, 10L, 11L, 12L, 13L));
+DataSet<Tuple7> source = env.fromCollection(
+ data, Types.TUPLE(Types.LONG, Types.LONG, Types.LONG, Types.LONG, Types.LONG, Types.LONG, Types.LONG));
+source.map(t -> {
+ Row row = new Row(7);
+ for (int i = 0; i < 7; i++) {
+ row.setField(i, t.getField(i));
+ }
+ return row;
+}).returns(rowTypeInfo).write(outputFormat, path);
+env.execute();
+```
diff --git a/flink-tsfile-connector/src/main/java/org/apache/iotdb/flink/tsfile/RowTSRecordConverter.java b/flink-tsfile-connector/src/main/java/org/apache/iotdb/flink/tsfile/RowTSRecordConverter.java
new file mode 100644
index 0000000..238085a
--- /dev/null
+++ b/flink-tsfile-connector/src/main/java/org/apache/iotdb/flink/tsfile/RowTSRecordConverter.java
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.flink.tsfile;
+
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Collector;
+import org.apache.iotdb.tsfile.common.constant.QueryConstant;
+import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
+import org.apache.iotdb.tsfile.utils.Binary;
+import org.apache.iotdb.tsfile.write.record.TSRecord;
+import org.apache.iotdb.tsfile.write.record.datapoint.BooleanDataPoint;
+import org.apache.iotdb.tsfile.write.record.datapoint.DataPoint;
+import org.apache.iotdb.tsfile.write.record.datapoint.DoubleDataPoint;
+import org.apache.iotdb.tsfile.write.record.datapoint.FloatDataPoint;
+import org.apache.iotdb.tsfile.write.record.datapoint.IntDataPoint;
+import org.apache.iotdb.tsfile.write.record.datapoint.LongDataPoint;
+import org.apache.iotdb.tsfile.write.record.datapoint.StringDataPoint;
+import org.apache.iotdb.tsfile.write.schema.Schema;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * The converter that convert a Row object to multiple TSRecord objects.
+ */
+public class RowTSRecordConverter implements TSRecordConverter<Row> {
+
+ private RowTypeInfo rowTypeInfo;
+ private transient TSRecord[] outputTemplate;
+ private transient int timeIndex = -1;
+ private transient int[] tsRecordIndexMapping;
+ private transient int[] dataPointIndexMapping;
+ private transient TSRecord[] reuse;
+
+ public RowTSRecordConverter(RowTypeInfo rowTypeInfo) {
+ this.rowTypeInfo = rowTypeInfo;
+ }
+
+ @Override
+ public void open(Schema schema) throws IOException {
+ this.tsRecordIndexMapping = new int[rowTypeInfo.getArity()];
+ this.dataPointIndexMapping = new int[rowTypeInfo.getArity()];
+ List<TSRecord> outputTemplateList = new ArrayList<>();
+
+ for (int i = 0; i < rowTypeInfo.getArity(); i++) {
+ String fieldName = rowTypeInfo.getFieldNames()[i];
+ if (QueryConstant.RESERVED_TIME.equals(fieldName)) {
+ timeIndex = i;
+ tsRecordIndexMapping[i] = -1;
+ dataPointIndexMapping[i] = -1;
+ continue;
+ }
+ String deviceId = fieldName.substring(0, fieldName.lastIndexOf("."));
+ String measurementId = fieldName.substring(fieldName.lastIndexOf(".") + 1);
+ int tsRecordIndex = outputTemplateList.stream()
+ .map(t -> t.deviceId).collect(Collectors.toList()).indexOf(deviceId);
+ if (tsRecordIndex < 0) {
+ outputTemplateList.add(new TSRecord(0, deviceId));
+ tsRecordIndex = outputTemplateList.size() - 1;
+ }
+ tsRecordIndexMapping[i] = tsRecordIndex;
+ TSRecord tsRecord = outputTemplateList.get(tsRecordIndex);
+ Class typeClass = rowTypeInfo.getFieldTypes()[i].getTypeClass();
+ if (typeClass == Boolean.class || typeClass == boolean.class) {
+ tsRecord.addTuple(new BooleanDataPoint(measurementId, false));
+ } else if (typeClass == Integer.class || typeClass == int.class) {
+ tsRecord.addTuple(new IntDataPoint(measurementId, 0));
+ } else if (typeClass == Long.class || typeClass == long.class) {
+ tsRecord.addTuple(new LongDataPoint(measurementId, 0));
+ } else if (typeClass == Float.class || typeClass == float.class) {
+ tsRecord.addTuple(new FloatDataPoint(measurementId, 0));
+ } else if (typeClass == Double.class || typeClass == double.class) {
+ tsRecord.addTuple(new DoubleDataPoint(measurementId, 0));
+ } else if (typeClass == String.class) {
+ tsRecord.addTuple(new StringDataPoint(measurementId, null));
+ } else {
+ throw new UnSupportedDataTypeException(typeClass.toString());
+ }
+ dataPointIndexMapping[i] = tsRecord.dataPointList.size() - 1;
+ }
+ outputTemplate = outputTemplateList.toArray(new TSRecord[0]);
+
+ reuse = new TSRecord[outputTemplate.length];
+ for (int i = 0; i < outputTemplate.length; i++) {
+ reuse[i] = new TSRecord(0, outputTemplate[i].deviceId);
+ }
+ }
+
+ @Override
+ public void covertAndCollect(Row input, Collector<TSRecord> collector) throws IOException {
+ long timestamp = (long) input.getField(timeIndex);
+ for (TSRecord tsRecord : reuse) {
+ tsRecord.dataPointList.clear();
+ }
+ for (int i = 0; i < input.getArity(); i++) {
+ if (i == timeIndex) {
+ continue;
+ }
+ TSRecord templateRecord = outputTemplate[tsRecordIndexMapping[i]];
+ DataPoint templateDataPoint = templateRecord.dataPointList.get(dataPointIndexMapping[i]);
+ Object o = input.getField(i);
+ if (o != null) {
+ Class typeClass = o.getClass();
+ if (typeClass == Boolean.class || typeClass == boolean.class) {
+ templateDataPoint.setBoolean((Boolean) o);
+ } else if (typeClass == Integer.class || typeClass == int.class) {
+ templateDataPoint.setInteger((Integer) o);
+ } else if (typeClass == Long.class || typeClass == long.class) {
+ templateDataPoint.setLong((Long) o);
+ } else if (typeClass == Float.class || typeClass == float.class) {
+ templateDataPoint.setFloat((Float) o);
+ } else if (typeClass == Double.class || typeClass == double.class) {
+ templateDataPoint.setDouble((Double) o);
+ } else if (typeClass == String.class) {
+ templateDataPoint.setString(Binary.valueOf((String) o));
+ } else {
+ throw new UnSupportedDataTypeException(typeClass.toString());
+ }
+ reuse[tsRecordIndexMapping[i]].addTuple(templateDataPoint);
+ }
+ }
+ for (TSRecord tsRecord : reuse) {
+ if (tsRecord.dataPointList.size() > 0) {
+ tsRecord.setTime(timestamp);
+ collector.collect(tsRecord);
+ }
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ outputTemplate = null;
+ timeIndex = -1;
+ tsRecordIndexMapping = null;
+ dataPointIndexMapping = null;
+ }
+}
diff --git a/flink-tsfile-connector/src/main/java/org/apache/iotdb/flink/tsfile/TSRecordConverter.java b/flink-tsfile-connector/src/main/java/org/apache/iotdb/flink/tsfile/TSRecordConverter.java
new file mode 100644
index 0000000..0d66425
--- /dev/null
+++ b/flink-tsfile-connector/src/main/java/org/apache/iotdb/flink/tsfile/TSRecordConverter.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.flink.tsfile;
+
+import org.apache.flink.util.Collector;
+import org.apache.iotdb.tsfile.write.record.TSRecord;
+import org.apache.iotdb.tsfile.write.schema.Schema;
+
+import java.io.IOException;
+import java.io.Serializable;
+
+public interface TSRecordConverter<T> extends Serializable {
+
+ void open(Schema schema) throws IOException;
+
+ void covertAndCollect(T input, Collector<TSRecord> collector) throws IOException;
+
+ void close() throws IOException;
+}
diff --git a/flink-tsfile-connector/src/main/java/org/apache/iotdb/flink/tsfile/TSRecordOutputFormat.java b/flink-tsfile-connector/src/main/java/org/apache/iotdb/flink/tsfile/TSRecordOutputFormat.java
new file mode 100644
index 0000000..9050da9
--- /dev/null
+++ b/flink-tsfile-connector/src/main/java/org/apache/iotdb/flink/tsfile/TSRecordOutputFormat.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.flink.tsfile;
+
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
+import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
+import org.apache.iotdb.tsfile.write.record.TSRecord;
+import org.apache.iotdb.tsfile.write.schema.Schema;
+
+import java.io.IOException;
+
+/**
+ * Output format that writes TsFiles by {@link TSRecord}. Users need to provide a {@link TSRecordConverter} used to
+ * convert the upstream data to {@link TSRecord}.
+ *
+ * @param <T> The input type of this output format.
+ */
+public class TSRecordOutputFormat<T> extends TsFileOutputFormat<T> {
+
+ private final TSRecordConverter<T> converter;
+
+ private transient TSRecordCollector tsRecordCollector = null;
+
+ public TSRecordOutputFormat(String path, Schema schema, TSRecordConverter<T> converter) {
+ this(path, schema, converter, null);
+ }
+
+ public TSRecordOutputFormat(Schema schema, TSRecordConverter<T> converter) {
+ super(null, schema, null);
+ this.converter = converter;
+ }
+
+ public TSRecordOutputFormat(String path, Schema schema, TSRecordConverter<T> converter, TSFileConfig config) {
+ super(path, schema, config);
+ this.converter = converter;
+ }
+
+ @Override
+ public void open(int taskNumber, int numTasks) throws IOException {
+ super.open(taskNumber, numTasks);
+ converter.open(schema);
+ tsRecordCollector = new TSRecordCollector();
+ }
+
+ @Override
+ public void close() throws IOException {
+ converter.close();
+ super.close();
+ }
+
+ @Override
+ public void writeRecord(T t) throws IOException {
+ try {
+ converter.covertAndCollect(t, tsRecordCollector);
+ } catch (FlinkRuntimeException e) {
+ throw new IOException(e.getCause());
+ }
+ }
+
+ private class TSRecordCollector implements Collector<TSRecord> {
+
+ @Override
+ public void collect(TSRecord tsRecord) {
+ try {
+ writer.write(tsRecord);
+ } catch (IOException | WriteProcessException e) {
+ throw new FlinkRuntimeException(e);
+ }
+ }
+
+ @Override
+ public void close() {
+
+ }
+ }
+
+ public TSRecordConverter<T> getConverter() {
+ return converter;
+ }
+}
\ No newline at end of file
diff --git a/flink-tsfile-connector/src/main/java/org/apache/iotdb/flink/tsfile/TsFileOutputFormat.java b/flink-tsfile-connector/src/main/java/org/apache/iotdb/flink/tsfile/TsFileOutputFormat.java
new file mode 100644
index 0000000..859e08f
--- /dev/null
+++ b/flink-tsfile-connector/src/main/java/org/apache/iotdb/flink/tsfile/TsFileOutputFormat.java
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.flink.tsfile;
+
+import org.apache.flink.api.common.io.FileOutputFormat;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.util.HadoopUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iotdb.flink.tsfile.util.TSFileConfigUtil;
+import org.apache.iotdb.hadoop.fileSystem.HDFSOutput;
+import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
+import org.apache.iotdb.tsfile.write.TsFileWriter;
+import org.apache.iotdb.tsfile.write.schema.Schema;
+import org.apache.iotdb.tsfile.write.writer.LocalTsFileOutput;
+import org.apache.iotdb.tsfile.write.writer.TsFileOutput;
+
+import javax.annotation.Nullable;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Optional;
+
+public abstract class TsFileOutputFormat<T> extends FileOutputFormat<T> {
+
+ protected Schema schema;
+ @Nullable
+ protected TSFileConfig config;
+
+ protected transient Configuration hadoopConf = null;
+ protected transient TsFileWriter writer = null;
+
+ public TsFileOutputFormat(String path, Schema schema, TSFileConfig config) {
+ super(path == null ? null : new Path(path));
+ this.schema = Preconditions.checkNotNull(schema);
+ this.config = config;
+ }
+
+ @Override
+ public void configure(org.apache.flink.configuration.Configuration flinkConfiguration) {
+ super.configure(flinkConfiguration);
+ hadoopConf = HadoopUtils.getHadoopConfiguration(flinkConfiguration);
+ }
+
+ @Override
+ public void open(int taskNumber, int numTasks) throws IOException {
+ super.open(taskNumber, numTasks);
+ if (config != null) {
+ TSFileConfigUtil.setGlobalTSFileConfig(config);
+ }
+ // Use TsFile API to write instead of FSDataOutputStream.
+ this.stream.close();
+ Path actualFilePath = getAcutalFilePath();
+ TsFileOutput out;
+ try {
+ if (actualFilePath.getFileSystem().isDistributedFS()) {
+ // HDFS
+ out = new HDFSOutput(
+ new org.apache.hadoop.fs.Path(new URI(actualFilePath.getPath())), hadoopConf, true);
+ } else {
+ // Local File System
+ out = new LocalTsFileOutput(new FileOutputStream(actualFilePath.getPath()));
+ }
+ } catch (URISyntaxException e) {
+ throw new RuntimeException(e);
+ }
+ writer = new TsFileWriter(out, schema);
+ }
+
+ @Override
+ public void close() throws IOException {
+ super.close();
+ writer.close();
+ writer = null;
+ }
+
+ @Override
+ protected String getDirectoryFileName(int taskNumber) {
+ return super.getDirectoryFileName(taskNumber) + ".tsfile";
+ }
+
+ protected Path getAcutalFilePath() {
+ try {
+ Field field = FileOutputFormat.class.getDeclaredField("actualFilePath");
+ field.setAccessible(true);
+ return (Path) field.get(this);
+ } catch (NoSuchFieldException | IllegalAccessException e) {
+ throw new RuntimeException("Get actual file path failed!", e);
+ }
+ }
+
+ public Schema getSchema() {
+ return schema;
+ }
+
+ public Optional<TSFileConfig> getConfig() {
+ return Optional.ofNullable(config);
+ }
+}
diff --git a/flink-tsfile-connector/src/test/java/org/apache/iotdb/flink/tsfile/RowTSRecordOutputFormatITCase.java b/flink-tsfile-connector/src/test/java/org/apache/iotdb/flink/tsfile/RowTSRecordOutputFormatITCase.java
new file mode 100644
index 0000000..3c7182e
--- /dev/null
+++ b/flink-tsfile-connector/src/test/java/org/apache/iotdb/flink/tsfile/RowTSRecordOutputFormatITCase.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.flink.tsfile;
+
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.types.Row;
+import org.junit.Test;
+
+import java.io.File;
+
+import static org.junit.Assert.assertArrayEquals;
+
+/**
+ * ITCases for TSRecordOutputFormat.
+ */
+public class RowTSRecordOutputFormatITCase extends RowTsFileOutputFormatTestBase {
+
+ @Test
+ public void testOutputFormat() throws Exception {
+ DataSet<Row> source = prepareDataSource();
+ String outputFilePath = tmpDir + File.separator + "test.tsfile";
+ TSRecordOutputFormat<Row> outputFormat = prepareTSRecordOutputFormat(outputFilePath);
+
+ source.output(outputFormat).setParallelism(1);
+ env.execute();
+
+ String[] actual = readTsFile(outputFilePath, paths);
+ String[] expected = {
+ "1,1.2,20,null,2.3,11,19",
+ "2,null,20,50,25.4,10,21",
+ "3,1.4,21,null,null,null,null",
+ "4,1.2,20,51,null,null,null",
+ "6,7.2,10,11,null,null,null",
+ "7,6.2,20,21,null,null,null",
+ "8,9.2,30,31,null,null,null"
+ };
+ assertArrayEquals(actual, expected);
+ }
+}
diff --git a/flink-tsfile-connector/src/test/java/org/apache/iotdb/flink/tsfile/RowTSRecordOutputFormatTest.java b/flink-tsfile-connector/src/test/java/org/apache/iotdb/flink/tsfile/RowTSRecordOutputFormatTest.java
new file mode 100644
index 0000000..6196540
--- /dev/null
+++ b/flink-tsfile-connector/src/test/java/org/apache/iotdb/flink/tsfile/RowTSRecordOutputFormatTest.java
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.flink.tsfile;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.types.Row;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for TSRecordOutputFormat
+ */
+public class RowTSRecordOutputFormatTest extends RowTsFileOutputFormatTestBase {
+
+ @Test
+ public void testWriteData() throws IOException {
+ String outputDirPath = tmpDir + File.separator + "testOutput";
+ new File(outputDirPath).mkdirs();
+ TSRecordOutputFormat<Row> outputFormat = prepareTSRecordOutputFormat(outputDirPath);
+
+ try {
+ outputFormat.configure(new Configuration());
+ outputFormat.open(0, 2);
+ List<Row> data = prepareData();
+ for (Row row : data) {
+ outputFormat.writeRecord(row);
+ }
+ } finally {
+ outputFormat.close();
+ }
+
+ String[] actual = readTsFile(outputDirPath + File.separator + "1.tsfile", paths);
+ String[] expected = {
+ "1,1.2,20,null,2.3,11,19",
+ "2,null,20,50,25.4,10,21",
+ "3,1.4,21,null,null,null,null",
+ "4,1.2,20,51,null,null,null",
+ "6,7.2,10,11,null,null,null",
+ "7,6.2,20,21,null,null,null",
+ "8,9.2,30,31,null,null,null"
+ };
+ assertArrayEquals(actual, expected);
+ }
+
+ @Test
+ public void testGetter() {
+ String outputFilePath = tmpDir + File.separator + "test.tsfile";
+ TSRecordOutputFormat<Row> outputFormat = prepareTSRecordOutputFormat(outputFilePath);
+
+ assertEquals(rowTSRecordConverter, outputFormat.getConverter());
+ assertEquals(schema, outputFormat.getSchema());
+ assertEquals(config, outputFormat.getConfig().get());
+ }
+}
diff --git a/flink-tsfile-connector/src/test/java/org/apache/iotdb/flink/tsfile/RowTsFileConnectorTestBase.java b/flink-tsfile-connector/src/test/java/org/apache/iotdb/flink/tsfile/RowTsFileConnectorTestBase.java
new file mode 100644
index 0000000..0ceb553
--- /dev/null
+++ b/flink-tsfile-connector/src/test/java/org/apache/iotdb/flink/tsfile/RowTsFileConnectorTestBase.java
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.flink.tsfile;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+
+import org.apache.flink.util.FileUtils;
+import org.apache.iotdb.flink.util.TsFileWriteUtil;
+
+import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
+import org.apache.iotdb.tsfile.common.constant.QueryConstant;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.junit.After;
+import org.junit.Before;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.List;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+/**
+ * Base class of the TsFile connector tests.
+ */
+public abstract class RowTsFileConnectorTestBase {
+
+ protected String tmpDir;
+ protected TSFileConfig config = new TSFileConfig();
+ protected String[] filedNames = {
+ QueryConstant.RESERVED_TIME,
+ "device_1.sensor_1",
+ "device_1.sensor_2",
+ "device_1.sensor_3",
+ "device_2.sensor_1",
+ "device_2.sensor_2",
+ "device_2.sensor_3"
+ };
+ protected TypeInformation[] typeInformations = new TypeInformation[] {
+ Types.LONG,
+ Types.FLOAT,
+ Types.INT,
+ Types.INT,
+ Types.FLOAT,
+ Types.INT,
+ Types.INT
+ };
+ protected List<Path> paths = Arrays.stream(filedNames)
+ .filter(s -> !s.equals(QueryConstant.RESERVED_TIME))
+ .map(Path::new)
+ .collect(Collectors.toList());
+ protected RowTypeInfo rowTypeInfo = new RowTypeInfo(typeInformations, filedNames);
+
+ @Before
+ public void prepareTempDirectory() throws Exception {
+ tmpDir = String.join(
+ File.separator,
+ TsFileWriteUtil.TMP_DIR,
+ UUID.randomUUID().toString());
+ new File(tmpDir).mkdirs();
+ config.setBatchSize(500);
+ }
+
+ @After
+ public void cleanTempDirectory() {
+ File tmpDirFile = new File(tmpDir);
+ FileUtils.deleteDirectoryQuietly(tmpDirFile);
+ }
+}
diff --git a/flink-tsfile-connector/src/test/java/org/apache/iotdb/flink/tsfile/RowTsFileInputFormatTestBase.java b/flink-tsfile-connector/src/test/java/org/apache/iotdb/flink/tsfile/RowTsFileInputFormatTestBase.java
index 9da84f7..48f5008 100644
--- a/flink-tsfile-connector/src/test/java/org/apache/iotdb/flink/tsfile/RowTsFileInputFormatTestBase.java
+++ b/flink-tsfile-connector/src/test/java/org/apache/iotdb/flink/tsfile/RowTsFileInputFormatTestBase.java
@@ -19,46 +19,25 @@
package org.apache.iotdb.flink.tsfile;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.common.typeinfo.Types;
-import org.apache.flink.api.java.typeutils.RowTypeInfo;
-
import org.apache.flink.types.Row;
import org.apache.iotdb.flink.util.TsFileWriteUtil;
-
-import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
-import org.apache.iotdb.tsfile.common.constant.QueryConstant;
-import org.apache.iotdb.tsfile.read.common.Path;
import org.apache.iotdb.tsfile.read.expression.QueryExpression;
-import org.junit.After;
import org.junit.Before;
import java.io.File;
-import java.util.Arrays;
-import java.util.List;
-import java.util.UUID;
-import java.util.stream.Collectors;
/**
- * Base class of the TsFileInputFormat tests.
+ * Base class for TsFileInputFormat tests.
*/
-public abstract class RowTsFileInputFormatTestBase {
+public abstract class RowTsFileInputFormatTestBase extends RowTsFileConnectorTestBase {
- protected String tmpDir;
protected String sourceTsFilePath1;
protected String sourceTsFilePath2;
- protected RowTypeInfo rowTypeInfo;
- protected TSFileConfig config;
- protected RowRowRecordParser parser;
- protected QueryExpression queryExpression;
+ protected RowRowRecordParser parser = RowRowRecordParser.create(rowTypeInfo, paths);
+ protected QueryExpression queryExpression = QueryExpression.create(paths, null);
@Before
public void prepareSourceTsFile() throws Exception {
- tmpDir = String.join(
- File.separator,
- TsFileWriteUtil.TMP_DIR,
- UUID.randomUUID().toString());
- new File(tmpDir).mkdirs();
sourceTsFilePath1 = String.join(
File.separator,
tmpDir, "source1.tsfile");
@@ -69,50 +48,7 @@ public abstract class RowTsFileInputFormatTestBase {
TsFileWriteUtil.create2(sourceTsFilePath2);
}
- @After
- public void removeSourceTsFile() {
- File sourceTsFile1 = new File(sourceTsFilePath1);
- if (sourceTsFile1.exists()) {
- sourceTsFile1.delete();
- }
- File sourceTsFile2 = new File(sourceTsFilePath2);
- if (sourceTsFile2.exists()) {
- sourceTsFile2.delete();
- }
- File tmpDirFile = new File(tmpDir);
- if (tmpDirFile.exists()) {
- tmpDirFile.delete();
- }
- }
-
protected TsFileInputFormat<Row> prepareInputFormat(String filePath) {
- String[] filedNames = {
- QueryConstant.RESERVED_TIME,
- "device_1.sensor_1",
- "device_1.sensor_2",
- "device_1.sensor_3",
- "device_2.sensor_1",
- "device_2.sensor_2",
- "device_2.sensor_3"
- };
- TypeInformation[] typeInformations = new TypeInformation[] {
- Types.LONG,
- Types.FLOAT,
- Types.INT,
- Types.INT,
- Types.FLOAT,
- Types.INT,
- Types.INT
- };
- List<Path> paths = Arrays.stream(filedNames)
- .filter(s -> !s.equals(QueryConstant.RESERVED_TIME))
- .map(Path::new)
- .collect(Collectors.toList());
- config = new TSFileConfig();
- config.setBatchSize(500);
- rowTypeInfo = new RowTypeInfo(typeInformations, filedNames);
- queryExpression = QueryExpression.create(paths, null);
- parser = RowRowRecordParser.create(rowTypeInfo, queryExpression.getSelectedSeries());
return new TsFileInputFormat<>(filePath, queryExpression, parser, config);
}
}
diff --git a/flink-tsfile-connector/src/test/java/org/apache/iotdb/flink/tsfile/RowTsFileOutputFormatTestBase.java b/flink-tsfile-connector/src/test/java/org/apache/iotdb/flink/tsfile/RowTsFileOutputFormatTestBase.java
new file mode 100644
index 0000000..4c993df
--- /dev/null
+++ b/flink-tsfile-connector/src/test/java/org/apache/iotdb/flink/tsfile/RowTsFileOutputFormatTestBase.java
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.flink.tsfile;
+
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.tuple.Tuple7;
+import org.apache.flink.types.Row;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.read.ReadOnlyTsFile;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.expression.QueryExpression;
+import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.iotdb.tsfile.write.schema.Schema;
+import org.junit.Before;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.iotdb.flink.util.TsFileWriteUtil.DEFAULT_TEMPLATE;
+
+/**
+ * Base class for TsFileOutputFormat tests.
+ */
+public abstract class RowTsFileOutputFormatTestBase extends RowTsFileConnectorTestBase {
+
+ protected ExecutionEnvironment env;
+ protected RowTSRecordConverter rowTSRecordConverter;
+ protected Schema schema;
+
+ @Before
+ public void prepareEnv() {
+ env = ExecutionEnvironment.getExecutionEnvironment();
+ }
+
+ protected TSRecordOutputFormat<Row> prepareTSRecordOutputFormat(String path) {
+ schema = new Schema();
+ schema.extendTemplate(
+ DEFAULT_TEMPLATE, new MeasurementSchema("sensor_1", TSDataType.FLOAT, TSEncoding.RLE));
+ schema.extendTemplate(
+ DEFAULT_TEMPLATE, new MeasurementSchema("sensor_2", TSDataType.INT32, TSEncoding.TS_2DIFF));
+ schema.extendTemplate(
+ DEFAULT_TEMPLATE, new MeasurementSchema("sensor_3", TSDataType.INT32, TSEncoding.TS_2DIFF));
+ rowTSRecordConverter = new RowTSRecordConverter(rowTypeInfo);
+ return new TSRecordOutputFormat<>(path, schema, rowTSRecordConverter, config);
+ }
+
+ protected List<Row> prepareData() {
+ List<Tuple7> tuples = new ArrayList<>(7);
+ tuples.add(new Tuple7(1L, 1.2f, 20, null, 2.3f, 11, 19));
+ tuples.add(new Tuple7(2L, null, 20, 50, 25.4f, 10, 21));
+ tuples.add(new Tuple7(3L, 1.4f, 21, null, null, null, null));
+ tuples.add(new Tuple7(4L, 1.2f, 20, 51, null, null, null));
+ tuples.add(new Tuple7(6L, 7.2f, 10, 11, null, null, null));
+ tuples.add(new Tuple7(7L, 6.2f, 20, 21, null, null, null));
+ tuples.add(new Tuple7(8L, 9.2f, 30, 31, null, null, null));
+
+ return tuples.stream().map(t -> {
+ Row row = new Row(7);
+ for (int i = 0; i < 7; i++) {
+ row.setField(i, t.getField(i));
+ }
+ return row;
+ }).collect(Collectors.toList());
+ }
+
+ protected DataSet<Row> prepareDataSource() {
+ List<Tuple7> input = new ArrayList<>(7);
+ input.add(new Tuple7(1L, 1.2f, 20, null, 2.3f, 11, 19));
+ input.add(new Tuple7(2L, null, 20, 50, 25.4f, 10, 21));
+ input.add(new Tuple7(3L, 1.4f, 21, null, null, null, null));
+ input.add(new Tuple7(4L, 1.2f, 20, 51, null, null, null));
+ input.add(new Tuple7(6L, 7.2f, 10, 11, null, null, null));
+ input.add(new Tuple7(7L, 6.2f, 20, 21, null, null, null));
+ input.add(new Tuple7(8L, 9.2f, 30, 31, null, null, null));
+ return env.fromCollection(prepareData(), rowTypeInfo);
+ }
+
+ protected String[] readTsFile(String tsFilePath, List<Path> paths) throws IOException {
+ QueryExpression expression = QueryExpression.create(paths, null);
+ TsFileSequenceReader reader = new TsFileSequenceReader(tsFilePath);
+ ReadOnlyTsFile readTsFile = new ReadOnlyTsFile(reader);
+ QueryDataSet queryDataSet = readTsFile.query(expression);
+ List<String> result = new ArrayList<>();
+ while (queryDataSet.hasNext()) {
+ RowRecord rowRecord = queryDataSet.next();
+ String row = rowRecord.getFields().stream()
+ .map(f -> f == null ? "null" : f.getStringValue())
+ .collect(Collectors.joining(","));
+ result.add(rowRecord.getTimestamp() + "," + row);
+ }
+ return result.toArray(new String[0]);
+ }
+}
diff --git a/flink-tsfile-connector/src/test/java/org/apache/iotdb/flink/util/TsFileWriteUtil.java b/flink-tsfile-connector/src/test/java/org/apache/iotdb/flink/util/TsFileWriteUtil.java
index 21d9ab7..7c3b8e8 100644
--- a/flink-tsfile-connector/src/test/java/org/apache/iotdb/flink/util/TsFileWriteUtil.java
+++ b/flink-tsfile-connector/src/test/java/org/apache/iotdb/flink/util/TsFileWriteUtil.java
@@ -36,7 +36,7 @@ import org.apache.iotdb.tsfile.write.schema.Schema;
public class TsFileWriteUtil {
public static final String TMP_DIR = "target";
- private static final String DEFAULT_TEMPLATE = "template";
+ public static final String DEFAULT_TEMPLATE = "template";
public static void create1(String tsfilePath) throws Exception {
File f = new File(tsfilePath);
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/schema/Schema.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/schema/Schema.java
index a1cb556..128b64f 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/schema/Schema.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/schema/Schema.java
@@ -20,6 +20,8 @@ package org.apache.iotdb.tsfile.write.schema;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.Path;
+
+import java.io.Serializable;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
@@ -28,7 +30,7 @@ import java.util.Map;
* The schema of timeseries that exist in this file. The deviceTemplates is a simplified manner to
* batch create schema of timeseries.
*/
-public class Schema {
+public class Schema implements Serializable {
/**
* Path (device + measurement) -> measurementSchema By default, use the LinkedHashMap to store the