创建maven工程

略

修改pom.xml

<repositories>
    <!-- 发布 CDH 的公司提供的maven仓库 -->
    <repository>
        <id>cloudera</id>
        <url>https://repository.cloudera.com/artifactory/cloudera-repos</url>
        <releases>
            <enabled>true</enabled>
        </releases>
        <snapshots>
            <enabled>true</enabled>
        </snapshots>
    </repository>
</repositories>
<properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
    <flink.version>1.12.0</flink.version>

</properties>

<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-planner-blink_2.12</artifactId>
        <version>${flink.version}</version>
    </dependency>

    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients_2.12</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-api</artifactId>
        <version>1.7.30</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>1.7.30</version>
    </dependency>
    <dependency>
        <groupId>log4j</groupId>
        <artifactId>log4j</artifactId>
        <version>1.2.17</version>
    </dependency>
</dependencies>

log4j.properties

不加会有警告提示

在 resources 目录下创建 log4j.properties，内容如下：

### 设置 ###
log4j.rootLogger = debug, debugFile, errorFile

### 输出debug级别以上的日志到：src/logs/debug.log ###
log4j.appender.debugFile = org.apache.log4j.DailyRollingFileAppender
log4j.appender.debugFile.File = debug.log
log4j.appender.debugFile.Append = true
log4j.appender.debugFile.Threshold = debug
log4j.appender.debugFile.layout = org.apache.log4j.PatternLayout
log4j.appender.debugFile.layout.ConversionPattern = %-d{yyyy-MM-dd HH:mm:ss}  [ %t:%r ] - [ %p ]  %n%m%n

### 输出error级别以上的日志到：src/logs/error.log ###
log4j.appender.errorFile = org.apache.log4j.DailyRollingFileAppender
log4j.appender.errorFile.File = error.log
log4j.appender.errorFile.Append = true
log4j.appender.errorFile.Threshold = error
log4j.appender.errorFile.layout = org.apache.log4j.PatternLayout
log4j.appender.errorFile.layout.ConversionPattern = %-d{yyyy-MM-dd HH:mm:ss}  [ %t:%r ] - [ %p ]  %n%m%n

java

查询所有列

package test;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.ArrayList;
import java.util.List;
import static org.apache.flink.table.api.Expressions.$;

import static org.apache.flink.table.api.Expressions.$;

public class TestTable {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//自动模式，根据数据源自行判断

        //构建数据
        DataStream ds=env.fromCollection(getData());

        //构建Table环境
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment stEnv = StreamTableEnvironment.create(env, bsSettings);

        //根据ds转换成table
        Table table=stEnv.fromDataStream(ds,$("name"),$("price"),$("username"));
        System.out.println("表名："+table);
        //执行查询
        Table res=stEnv.sqlQuery("select * from "+table+" where price>60");
        System.out.println("Schema:");
        //打印Schema
        res.printSchema();

        //需要将结果转换成DataStream，才能打印结果
        /*
        将计算后的数据，在ds原数据基础上更新
        true：表示更新
        false：表示删除
         */

        DataStream<Tuple2<Boolean, Order>> dsRes = stEnv.toRetractStream(res, Order.class);
        dsRes.print();
        //将计算后的数据，append到ds，支持简单查询
//        DataStream<Order> dsRes2 = stEnv.toAppendStream(res, Order.class);

        env.execute();
    }

    public static List getData(){
        List<Order> list=new ArrayList();
        list.add(new Order("java从入门到精通",50,"李雷"));
        list.add(new Order("hadoop从入门到精通",80,"韩梅梅"));
        list.add(new Order("flink从入门到精通",70,"韩梅梅"));

        return list;
    }
    public static class Order {
        String name;
        double price;
        String username;

        //注意：需要无参构造方法，否则flink会报错
        public Order() {
        }

        public Order(String name, double price, String username) {
            this.name = name;
            this.price = price;
            this.username = username;
        }

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public double getPrice() {
            return price;
        }

        public void setPrice(double price) {
            this.price = price;
        }

        public String getUsername() {
            return username;
        }

        public void setUsername(String username) {
            this.username = username;
        }

        //增加toString() 方法，方便打印查看结果
        @Override
        public String toString() {
            return "Order{" +
                    "name='" + name + '\'' +
                    ", price=" + price +
                    ", username='" + username + '\'' +
                    '}';
        }
    }
}

注意：

Order 类必须有 无参构造方法，否则flink会报错

不查询所有列

如果查询2列，需要用与之匹配的 Bean，否则用 Row 接收

原文出处：/show_1IX23DvPepeM.html