flinksql 表怎么讀取外部文件

這篇文章主要介紹“flinksql 表怎么讀取外部文件”，在日常操作中，相信很多人在flinksql 表怎么讀取外部文件問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”flinksql 表怎么讀取外部文件”的疑惑有所幫助！接下來，請跟著小編一起來學習吧！

1、TableEnvironment 可以注冊目錄 Catalog， 可以基于catalog 注冊表

2、表 是由一個標識符 來指定的， 由三部分組成：catalog、 數據庫名、對象名

3、表可以是常規的，也可以是虛擬的（視圖）

4、常規表一般可以用來描述外部數據， 比如文件、數據庫或消息隊列的數據，也可以直接從datastream轉換過來

5、視圖可以從現有的表中創建，通常是table api 或者 sql 查詢的一個結果集

代碼：

package com.jd.data;

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;
import org.apache.flink.types.Row;
import scala.Tuple3;

public class TableAipDemo03 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);


//        1、創建表執行環節
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String path = "/Users/liuhaijing/Desktop/flinktestword/aaa.txt";
        tableEnv.connect(new FileSystem().path(path))
                .withFormat(new OldCsv()) // 定義格式化方法
                .withSchema(new Schema().field("a", DataTypes.STRING())  // 定義表的結構
                        .field("b", DataTypes.STRING())
                        .field("c", DataTypes.STRING())

                )
                .inAppendMode()
                .createTemporaryTable("xxx");

        Table xxx = tableEnv.from("xxx");

        xxx.printSchema();

        tableEnv.toAppendStream(xxx,  Row.class ).print();

        env.execute("job");


    }
}

到此，關于“flinksql 表怎么讀取外部文件”的學習就結束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續學習更多相關知識，請繼續關注億速云網站，小編會繼續努力為大家帶來更多實用的文章！