Flink IoTDB 连接器

IoTDB 与 Apache FlinkFlink-IoTDB - 图1 (opens new window) 的集成。此模块包含了 iotdb sink，允许 flink job 将时序数据写入 IoTDB。

IoTDBSink

使用 IoTDBSink ，您需要定义一个 IoTDBOptions 和一个 IoTSerializationSchema 实例。 IoTDBSink 默认每次发送一个数据，可以通过调用 withBatchSize(int) 进行调整。

示例

该示例演示了如下从一个 Flink job 中发送数据到 IoTDB server 的场景：

  • 一个模拟的 Source SensorSource 每秒钟产生一个数据点。

  • Flink 使用 IoTDBSink 消费产生的数据并写入 IoTDB 。

    1. import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
    2. import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
    3. import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
    5. import com.google.common.collect.Lists;
    6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    7. import org.apache.flink.streaming.api.functions.source.SourceFunction;
    9. import java.security.SecureRandom;
    10. import java.util.HashMap;
    11. import java.util.Map;
    12. import java.util.Random;
    14. public class FlinkIoTDBSink {
    15.   public static void main(String[] args) throws Exception {
    16.     // run the flink job on local mini cluster
    17.     StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    19.     IoTDBOptions options = new IoTDBOptions();
    20.     options.setHost("127.0.0.1");
    21.     options.setPort(6667);
    22.     options.setUser("root");
    23.     options.setPassword("root");
    24.     options.setStorageGroup("root.sg");
    26.     // If the server enables auto_create_schema, then we do not need to register all timeseries
    27.     // here.
    28.     options.setTimeseriesOptionList(
    29.         Lists.newArrayList(
    30.             new IoTDBOptions.TimeseriesOption(
    31.                 "root.sg.d1.s1", TSDataType.DOUBLE, TSEncoding.GORILLA, CompressionType.SNAPPY)));
    33.     IoTSerializationSchema serializationSchema = new DefaultIoTSerializationSchema();
    34.     IoTDBSink ioTDBSink =
    35.         new IoTDBSink(options, serializationSchema)
    36.             // enable batching
    37.             .withBatchSize(10)
    38.             // how many connectons to the server will be created for each parallelism
    39.             .withSessionPoolSize(3);
    41.     env.addSource(new SensorSource())
    42.         .name("sensor-source")
    43.         .setParallelism(1)
    44.         .addSink(ioTDBSink)
    45.         .name("iotdb-sink");
    47.     env.execute("iotdb-flink-example");
    48.   }
    50.   private static class SensorSource implements SourceFunction<Map<String, String>> {
    51.     boolean running = true;
    52.     Random random = new SecureRandom();
    54.     @Override
    55.     public void run(SourceContext context) throws Exception {
    56.       while (running) {
    57.         Map<String, String> tuple = new HashMap();
    58.         tuple.put("device", "root.sg.d1");
    59.         tuple.put("timestamp", String.valueOf(System.currentTimeMillis()));
    60.         tuple.put("measurements", "s1");
    61.         tuple.put("types", "DOUBLE");
    62.         tuple.put("values", String.valueOf(random.nextDouble()));
    64.         context.collect(tuple);
    65.         Thread.sleep(1000);
    66.       }
    67.     }
    69.     @Override
    70.     public void cancel() {
    71.       running = false;
    72.     }
    73.   }
    74. }

运行方法

  • 启动 IoTDB server
  • 运行 org.apache.iotdb.flink.FlinkIoTDBSink.java 将 Flink job 运行在本地的集群上。