How does flinksql write data to a file

This article mainly explains "how flinksql writes data to a file". Interested friends may wish to have a look at it. The method introduced in this paper is simple, fast and practical. Let's let the editor take you to learn how flinksql writes data to a file.

Package com.jd.dataoutput;import com.jd.data.SensorReading;import org.apache.flink.api.common.functions.MapFunction;import org.apache.flink.streaming.api.datastream.DataStreamSource;import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.table.api.DataTypes;import org.apache.flink.table.api.Table;import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;import org.apache.flink.table.descriptors.FileSystem Import org.apache.flink.table.descriptors.OldCsv;import org.apache.flink.table.descriptors.Schema;public class FlinkSqlOutputFile {public static void main (String [] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment (); env.setParallelism (1); DataStreamSource stream = env.readTextFile ("/ Users/liuhaijing/Desktop/flinktestword/aaa.txt"); / / DataStreamSource stream = env.socketTextStream ("localhost", 8888) SingleOutputStreamOperator map = stream.map (new MapFunction () {public SensorReading map (String s) throws Exception {String [] split = s.split (","); return new SensorReading (split [0], split [1], split [2]);}}); StreamTableEnvironment tableEnv = StreamTableEnvironment.create (env) / / use table api Table table = tableEnv.fromDataStream (map); / / table.printSchema (); Table select = table.select ("test b"); / / select.printSchema (); / / use sql api// tableEnv.createTemporaryView ("test", map); / / Table select = tableEnv.sqlQuery ("select aline b from test"); / / select.printSchema () / / DataStream sensorReading2DataStream = tableEnv.toAppendStream (select, SensorReading2.class); / / sensorReading2DataStream.map (new MapFunction () {/ / @ Override// public Object map (SensorReading2 value) throws Exception {/ / System.out.println (value.a+ "" + value.b); / / return null;//} / /}) / / tableEnv.connect (new FileSystem (). Path ("/ Users/liuhaijing/IdeaProjects/haijing3/spark/flinksqldemo/output/out.txt")) / / .withFormat (new Csv ()) / / .withSchema (/ / new Schema () / / .field ("a") DataTypes.STRING () / / .field ("b", DataTypes.STRING ()) / / .inAppendMode () / / .createTemporaryTable ("outputTable") / / select.insertInto ("outputTable") TableEnv.connect (new FileSystem (). Path ("/ Users/liuhaijing/IdeaProjects/haijing3/spark/flinksqldemo/output/out.txt")) .withFormat (new OldCsv ()) .withSchema (new Schema () .field ("a", DataTypes.STRING () .inAppendMode () .createTemporaryTable ("outputTable") Select.insertInto ("outputTable"); env.execute ();}} at this point, I believe you have a better understanding of "how flinksql writes data to a file". You might as well do it in practice! Here is the website, more related content can enter the relevant channels to inquire, follow us, continue to learn!