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;

  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. IoTDBOptions options = new IoTDBOptions();

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

  17. options.setStorageGroup("root.sg");

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

  19. // here.
  20. options.setTimeseriesOptionList(
  21. Lists.newArrayList(
  22. new IoTDBOptions.TimeseriesOption(

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

  24. IoTSerializationSchema serializationSchema = new DefaultIoTSerializationSchema();

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

  30. .withSessionPoolSize(3);

  31. env.addSource(new SensorSource())

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

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

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

  37. }

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

  39. boolean running = true;

  40. Random random = new SecureRandom();

  41. @Override

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

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

  50. context.collect(tuple);

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

  53. }

  54. @Override

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

运行方法

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