To use the , you need construct an instance of it by specifying IoTDBSinkOptions and IoTSerializationSchema instances. The IoTDBSink send only one event after another by default, but you can change to batch by invoking withBatchSize(int).

Example

• A simulated Source SensorSource generates data points per 1 second.
• Flink uses IoTDBSink to consume the generated data points and write the data into IoTDB.

It is noteworthy that to use IoTDBSink, schema auto-creation in IoTDB should be enabled.

Usage

• Launch the IoTDB server.
• Run org.apache.iotdb.flink.FlinkIoTDBSink.java to run the flink job on local mini cluster.

IoTDBSource

Example

This example shows a case where data are read from IoTDB.

import org.apache.iotdb.flink.options.IoTDBSourceOptions;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.session.Session;
import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
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.RowRecord;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.util.ArrayList;
import java.util.List;
public class FlinkIoTDBSource {
  static final String ROOT_SG1_D1_S1 = "root.sg1.d1.s1";
  public static void main(String[] args) throws Exception {
    prepareData();
    // run the flink job on local mini cluster
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    IoTDBSourceOptions ioTDBSourceOptions =
        new IoTDBSourceOptions("127.0.0.1", 6667, "root", "root",
            "select s1 from " + ROOT_SG1_D1 + " align by device");
    env.addSource(
        new IoTDBSource<RowRecord>(ioTDBSourceOptions) {
          @Override
          public RowRecord convert(RowRecord rowRecord) {
            return rowRecord;
          }
        })
        .name("sensor-source")
        .print()
        .setParallelism(2);
    env.execute();
  }
  /**
   * Write some data to IoTDB
   */
  private static void prepareData() throws IoTDBConnectionException, StatementExecutionException {
    try {
      session.setStorageGroup("root.sg1");
      if (!session.checkTimeseriesExists(ROOT_SG1_D1_S1)) {
        session.createTimeseries(
            ROOT_SG1_D1_S1, TSDataType.INT64, TSEncoding.RLE, CompressionType.SNAPPY);
        List<String> measurements = new ArrayList<>();
        List<TSDataType> types = new ArrayList<>();
        measurements.add("s1");
        measurements.add("s2");
        measurements.add("s3");
        types.add(TSDataType.INT64);
        types.add(TSDataType.INT64);
        types.add(TSDataType.INT64);
        for (long time = 0; time < 100; time++) {
          List<Object> values = new ArrayList<>();
          values.add(1L);
          values.add(2L);
          values.add(3L);
          session.insertRecord(ROOT_SG1_D1, time, measurements, types, values);
        }
      }
    } catch (StatementExecutionException e) {
      if (e.getStatusCode() != TSStatusCode.PATH_ALREADY_EXIST_ERROR.getStatusCode()) {
        throw e;
      }
    }
  }
}

Usage