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 表怎么读取外部文件”的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注亿速云网站，小编会继续努力为大家带来更多实用的文章！