flinksql怎么將數據寫入到文件中

本篇內容主要講解“flinksql怎么將數據寫入到文件中”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓小編來帶大家學習“flinksql怎么將數據寫入到文件中”吧!

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<String> stream = env.readTextFile("/Users/liuhaijing/Desktop/flinktestword/aaa.txt");
//        DataStreamSource<String> stream = env.socketTextStream("localhost", 8888);

        SingleOutputStreamOperator<SensorReading> map = stream.map(new MapFunction<String, SensorReading>() {

            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);
//        使用 table api
        Table table = tableEnv.fromDataStream(map);
//        table.printSchema();
        Table select = table.select("a,b");
//        select.printSchema();

//        使用 sql api
//        tableEnv.createTemporaryView("test", map);
//        Table select = tableEnv.sqlQuery(" select a, b from test");
//        select.printSchema();


//        DataStream<SensorReading2> sensorReading2DataStream = tableEnv.toAppendStream(select, SensorReading2.class);
//        sensorReading2DataStream.map(new MapFunction<SensorReading2, Object>() {
//            @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();




    }
}

到此，相信大家對“flinksql怎么將數據寫入到文件中”有了更深的了解，不妨來實際操作一番吧！這里是億速云網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續學習！