Apache Flink

IoTDB 与 Apache Flink 的集成。此模块包含了 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.flink.options.IoTDBSinkOptions; 2. import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType; 3. import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; 4. import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;

  1. import com.google.common.collect.Lists;
  2. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

  3. import org.apache.flink.streaming.api.functions.source.SourceFunction;

  4. import java.security.SecureRandom;

  5. import java.util.HashMap;
  6. import java.util.Map;

  7. import java.util.Random;

  8. public class FlinkIoTDBSink {

  9. public static void main(String[] args) throws Exception {
  10. // run the flink job on local mini cluster

  11. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

  12. IoTDBSinkOptions options = new IoTDBSinkOptions();

  13. options.setHost("127.0.0.1");
  14. options.setPort(6667);
  15. options.setUser("root");

  16. options.setPassword("root");

  17. // If the server enables auto_create_schema, then we do not need to register all timeseries

  18. // here.
  19. options.setTimeseriesOptionList(
  20. Lists.newArrayList(
  21. new IoTDBSinkOptions.TimeseriesOption(

  22. "root.sg.d1.s1", TSDataType.DOUBLE, TSEncoding.GORILLA, CompressionType.SNAPPY)));

  23. IoTSerializationSchema serializationSchema = new DefaultIoTSerializationSchema();

  24. IoTDBSink ioTDBSink =
  25. new IoTDBSink(options, serializationSchema)
  26. // enable batching
  27. .withBatchSize(10)
  28. // how many connections to the server will be created for each parallelism

  29. .withSessionPoolSize(3);

  30. env.addSource(new SensorSource())

  31. .name("sensor-source")
  32. .setParallelism(1)
  33. .addSink(ioTDBSink)

  34. .name("iotdb-sink");

  35. env.execute("iotdb-flink-example");

  36. }

  37. private static class SensorSource implements SourceFunction<Map> {

  38. boolean running = true;

  39. Random random = new SecureRandom();

  40. @Override

  41. public void run(SourceContext context) throws Exception {
  42. while (running) {
  43. Map tuple = new HashMap();
  44. tuple.put("device", "root.sg.d1");
  45. tuple.put("timestamp", String.valueOf(System.currentTimeMillis()));
  46. tuple.put("measurements", "s1");
  47. tuple.put("types", "DOUBLE");

  48. tuple.put("values", String.valueOf(random.nextDouble()));

  49. context.collect(tuple);

  50. Thread.sleep(1000);
  51. }

  52. }

  53. @Override

  54. public void cancel() {
  55. running = false;
  56. }
  57. }
  58. } ```

运行方法

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