【技术干货】代码示例：使用 Apache Flink 连接 TDengine

package com.taosdata.flink;import org.apache.flink.configuration.Configuration;import org.apache.flink.streaming.api.functions.source.RichSourceFunction;import com.taosdata.model.Sensor;import java.sql.*;import java.util.Properties;
public class SourceFromTDengine extends RichSourceFunction<Sensor> {    Statement statement;    private Connection connection;    private String property;    public SourceFromTDengine(){        super();    }
    @Override    public void open(Configuration parameters) throws Exception {        super.open(parameters);        String driver = "com.taosdata.jdbc.rs.RestfulDriver";        String host = "u05";        String username = "root";        String password = "taosdata";        String prop = System.getProperty("java.library.path");        Logger LOG = LoggerFactory.getLogger(SourceFromTDengine.class);        LOG.info("java.library.path:{}", prop);        System.out.println(prop);        Class.forName( driver );        Properties properties = new Properties();        connection = DriverManager.getConnection("jdbc:TAOS-RS://" + host + ":6041/tt" + "?user=root&password=taosdata"                , properties);        statement = connection.createStatement();    }
    @Override    public void close() throws Exception {        super.close();        if (connection != null) {            connection.close();        }        if (statement != null) {            statement.close();        }    }
    @Override    public void run(SourceContext<Sensor> sourceContext) throws Exception {        try {            String sql = "select * from tt.meters";            ResultSet resultSet = statement.executeQuery(sql);            while (resultSet.next()) {                Sensor sensor = new Sensor( resultSet.getLong(1),                        resultSet.getInt( "vol" ),                        resultSet.getFloat( "current" ),                        resultSet.getString( "location" ).trim());                sourceContext.collect( sensor );            }        } catch (Exception e) {            e.printStackTrace();        }    }
    @Override    public void cancel() {
    }}

package com.taosdata.flink;
import org.apache.flink.configuration.Configuration;import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;import com.taosdata.model.Sensor;import java.sql.*;import java.util.Properties;

public class SinkToTDengine extends RichSinkFunction<Sensor> {    Statement statement;    private Connection connection;
    @Override    public void open(Configuration parameters) throws Exception {        super.open(parameters);        String driver = "com.taosdata.jdbc.rs.RestfulDriver";        String host = "TAOS-FQDN";        String username = "root";        String password = "taosdata";        String prop = System.getProperty("java.library.path");        System.out.println(prop);        Class.forName( driver );        Properties properties = new Properties();        connection = DriverManager.getConnection("jdbc:TAOS-RS://" + host + ":6041/tt" + "?user=root&password=taosdata"                , properties);        statement = connection.createStatement();
    }
    @Override    public void close() throws Exception {        super.close();        if (connection != null) {            connection.close();        }        if (statement != null) {            statement.close();        }    }
    @Override    public void invoke(Sensor sensor, Context context) throws Exception {        try {            String sql = String.format("insert into sinktest.%s using sinktest.meters tags('%s') values(%d,%d,%f)",                                 sensor.getLocation(),                                sensor.getLocation(),                                sensor.getTs(),                                sensor.getVal(),                                sensor.getCurrent()                                );            statement.executeUpdate(sql);
        } catch (Exception e) {            e.printStackTrace();        }    }}

package com.taosdata.model;
public class Sensor {
    public long ts;    public int val;    public float current;    public String location;
    public Sensor() {
    }
    public Sensor(long ts, int val, float current, String location) {        this.ts = ts;        this.val = val;        this.current = current;        this.location = location;    }
    public long getTs() {        return ts;    }
    public void setTs(long ts) {        this.ts = ts;    }
    public int getVal() {        return val;    }
    public void setVal(int val) {        this.val = val;    }
    public float getCurrent() {        return current;    }
    public void setCurrent(float current) {        this.current = current;    }
    public String getLocation() {        return location;    }
    public void setLocation(String location) {        this.location = location;    }
    @Override    public String toString() {        return "Sensor{" +                "ts=" + ts +                ", val=" + val +                ", current=" + current +                ", location='" + location + '\'' +                '}';    }}

package com.taosdata;
import org.apache.flink.streaming.api.datastream.DataStreamSource;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.api.common.functions.MapFunction;import org.apache.flink.streaming.api.datastream.DataStream;import com.taosdata.model.Sensor;
import org.slf4j.LoggerFactory;import org.slf4j.Logger;
public class ReadFromTDengine {    public static void main(String[] args) throws Exception {        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        DataStreamSource<Sensor> SensorList = env.addSource( new com.taosdata.flink.SourceFromTDengine() );        SensorList.print();        SensorList.addSink( new com.taosdata.flink.SinkToTDengine() );               env.execute();
    }}