flink1.12.x SQL：创建工程，从socket读取数据，实现查询（DataStreamSource转View）

作者：马育民 • • 阅读：10152

说明

文中案例，连接 socket ，从socket读取数据，实现查询

创建maven工程

略

maven

设置maven

pom.xml

关键依赖

在之前的依赖基础上，增加下面依赖：

<!--  table api、sql需要的依赖  -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-planner-blink_2.12</artifactId>
    <version>${flink.version}</version>
</dependency>

完整配置

<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>
    <!--java-->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_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>
    <!-- kafka Dependency -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka_2.12</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-files</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-runtime-web_2.12</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-planner-blink_2.12</artifactId>
        <version>${flink.version}</version>
    </dependency>
</dependencies>

log4j.properties

提示：不加会有警告提示

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

# DEBUG 级别，即：所有信息都会输出
log4j.rootLogger=WARN, stdout
# 配置 stdout，输出到控制台
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n

java

封装类

package p2socket;

public class Order {
    private String name;
    private Double price;
    private String username;

    // 要有无参构造方法
    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;
    }

    @Override
    public String toString() {
        return "Order{" +
                "name='" + name + '\'' +
                ", price=" + price +
                ", username='" + username + '\'' +
                '}';
    }
}

注意：要有无参构造方法

主类

package p2socket;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
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.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

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

public class C1SocketView {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//自动模式，根据数据源自行判断
        //构建Table环境
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment stEnv = StreamTableEnvironment.create(env, bsSettings);

        DataStreamSource<String> dss = env.socketTextStream("127.0.0.1", 9999, "\n");
        SingleOutputStreamOperator<Order> soso = dss.flatMap(new FlatMapFunction<String, Order>() {
            @Override
            public void flatMap(String value, Collector<Order> out) throws Exception {
                String[] arr = value.split(",");
                Double price = Double.parseDouble(arr[1]);
                Order order = new Order(arr[0], price, arr[2]);

                out.collect(order);
            }
        });
        // 创建临时视图，不要使用关键字，如：order
        stEnv.createTemporaryView("v_order",soso);
        // 也可以指定字段
        // stEnv.createTemporaryView("v_order",soso,$("name"),$("price"),$("username"));
        // 以封装类的成员变量名，作为 列名
        Table res=stEnv.sqlQuery("select * from v_order where price>60");
        System.out.println("Schema信息:");
        //打印Schema
        res.printSchema();
        /*
        将计算后的数据，append到ds，适用于普通查询
        分组查询，来一条记录，需要重新计算，并更新原结果， toAppendStream 方法不支持，会报错
         */
        DataStream<Order> dsRes = stEnv.toAppendStream(res, Order.class);
        dsRes.print();
        env.execute();
    }
}

注意：

  • 临时视图，不要出现关键字，如：order 等

测试

启动 socket 服务器

发送下面数据

java,50,lilei
hive,80,hanmeimei
hadoop,90,lilei
flink,70,lucy
mysql,100,lili
hive,100,lucy
html,120,lucy

执行结果

Schema信息:
root
 |-- name: STRING
 |-- price: DOUBLE
 |-- username: STRING

5> Order{name='java', price=80.0, username='lilei'}

原文出处：/show_1IX4E2CFdBMf.html

吉ICP备15003043号