# FlinkIoTDBSink样例程序
#### 功能简介
IoTDB与Flink的集成。此模块包含了iotdb sink，通过flink job将时序数据写入IoTDB。
#### 代码样例
该样例演示了从一个Flink job中发送数据到IoTDB server的场景。
- 一个模拟的Source SensorSource每秒钟产生一个数据点。
- Flink使用IoTDBSink消费产生数据并写入IoTDB。
其中在Session对象的参数里，设置IoTDBServer所在的节点IP、端口、用户名和密码。
![](https://support.huaweicloud.com/devg-lts-mrs/public_sys-resources/note_3.0-zh-cn.png)
- 待连接的IoTDBServer所在的节点IP地址，可通过登录FusionInsight Manager界面，选择"集群 \> 服务 \> IoTDB \> 实例"查看。

- RPC端口可通过登录FusionInsight Manager，选择"集群 \> 服务 \> IoTDB \> 配置 \> 全部配置"，搜索参数"IOTDB_SERVER_RPC_PORT"获得。
- 普通模式下，初始安装后IoTDB有一个默认用户：root，密码请参见[用户账号一览表](https://support.huaweicloud.com/usermanual-mrs/admin_guide_000239.html)章节获取。该用户为管理员用户，固定拥有所有权限，无法被赋予，无法被撤销权限，也无法被删除。
- 需在本地环境变量中设置环境变量认证用户名和认证用户密码，建议密文存放，使用时解密，确保安全。其中：
  - *认证用户名*为访问IoTDB的用户名。
  
  - *认证用户密码*为访问IoTDB的用户密码。
   
 
```
public class FlinkIoTDBSink {
  public static void main(String[] args) throws Exception {
    // run the flink job on local mini cluster
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    IoTDBSinkOptions options = new IoTDBSinkOptions();
    options.setHost("127.0.0.1");
    options.setPort(22260);
    options.setUser("用户名");
    options.setPassword("密码");
    // If the server enables auto_create_schema, then we do not need to register all timeseries
    // here.
    options.setTimeseriesOptionList(
            Lists.newArrayList(
                    new IoTDBSinkOptions.TimeseriesOption(
                            "root.sg.d1.s1", TSDataType.DOUBLE, TSEncoding.GORILLA, CompressionType.SNAPPY)));
    IoTSerializationSchema serializationSchema = new DefaultIoTSerializationSchema();
    IoTDBSink ioTDBSink =
        new IoTDBSink(options, serializationSchema)
            // enable batching
            .withBatchSize(10)
            // how many connections to the server will be created for each parallelism
            .withSessionPoolSize(3);
    env.addSource(new SensorSource())
        .name("sensor-source")
        .setParallelism(1)
        .addSink(ioTDBSink)
        .name("iotdb-sink");
    env.execute("iotdb-flink-example");
  }
  private static class SensorSource implements SourceFunction<Map<String, String>> {
    boolean running = true;
    Random random = new SecureRandom();
    @Override
    public void run(SourceContext context) throws Exception {
      while (running) {
        Map<String, String> tuple = new HashMap();
        tuple.put("device", "root.sg.d1");
        tuple.put("timestamp", String.valueOf(System.currentTimeMillis()));
        tuple.put("measurements", "s1");
        tuple.put("types", "DOUBLE");
        tuple.put("values", String.valueOf(random.nextDouble()));
        context.collect(tuple);
        Thread.sleep(1000);
      }
    }
    @Override
    public void cancel() {
      running = false;
    }
  }
}
```