通过定制Flink的Sink类，实现向vm时序数据库的数据写入。
Flink代码主函数条目
package example； importorg.Apache.flink.streaming.API.datastream.datastream； importorg.Apache.flink.streaming.API.environment.streamexecutionenvironment； import java.util.ArrayList； import java.util.List； /** * @author admin * p * using flink，writetovmuseinfluxdbprotocol */public class write2vmbyflink { publicstaticlistringdistiong i 10； I ) list.add('I )； } return list； } publicstaticvoidmain (stringbzdtnargs ) throws exception (finalstreamexecutionenvironmentenv=streamexecutionenvironment . datastream source.addsink (newvmsink ) ) ).name ) ) VMsink )； env.execute (； }} Sink类
package example； import com.influx db.client.influxdbclient； import com.influx db.client.influxdbclientfactory； import com.influx db.client.write API； import com.influx db.client.domain.write precision； importorg.Apache.flink.configuration.configuration； importorg.Apache.flink.streaming.API.functions.sink.richsinkfunction； import com.influx db.client.write.point； import java.time.Instant；/* * @ author admin */publicclassvmsinkextendsrichsinkfunctionstring { privatestring bucket=' flink '； private String org='galaxy '； 私有写入应用程序写入应用程序； @ overridepublicvoidopen (configuration parameters ) throwsexception(if ) null==writeAPI ) influxdbclientclient=influxdexdexdstint } @ overridepublicvoidinvoke (string value， 上下文上下文) throws exception (pointpoint=point.measurement ) (“sun”) ) ) ) )“flinkSink”).addfied integer.parseint(value ) *2).time ) instant.now )、write precision write API.write point (bucket、org、point )； } @Override public void close () throwsexception ) if ) null==writeAPI ) influxdbclientclient=influxdbclientfactory.cried } writeApi.flush (； }实际将数据写入虚拟机需要闪存。 这个示范例子只是初步的尝试。 如意煎饼p如意煎饼/pfzddtpmaven依赖
flink.version1. 14.0/flink.versiondependencygroupidorg.Apache.flink/groupidartifactidflink-Java/artifactidvink ependencydependencygroupidorg.Apache.flink/groupidartifactidflink-streaming-Java _ 2.11/artifactidvactifactidvaction ependencydependencygroupidorg.Apache.flink/groupidartifactidflink-clients _ 2.11/artifactidversion $ { flink.verss ependencydependencygroupidcom.influx db/groupidartifactidinfluxdb-client-Java/artifactidversion 2.0/version/如意煎饼pfzddt/p如意煎饼pfzddt/p如意煎饼pfzddt/p如意煎饼pfzddt/p如意煎饼PP丰富柠檬/p如意煎饼PP丰富柠檬/p