Shulou(Shulou.com)06/01 Report--

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!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.