练习 : Flink 批流一体

pom

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0modelVersion>

    <groupId>org.examplegroupId>
    <artifactId>test4-8artifactId>
    <version>1.0-SNAPSHOTversion>
    <properties>
        <project.build.sourceEncoding>UTF-8project.build.sourceEncoding>
        <maven.compiler.source>1.8maven.compiler.source>
        <maven.compiler.target>1.8maven.compiler.target>
        <flink.version>1.13.0flink.version>
        <hadoop.version>3.1.3hadoop.version>
        <scala.version>2.12scala.version>
        <scala.binary.version>2.11scala.binary.version>
    properties>

    <dependencies>

        
        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-coreartifactId>
            <version>${flink.version}version>
        dependency>
        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-javaartifactId>
            <version>${flink.version}version>
        dependency>
        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-streaming-java_${scala.version}artifactId>
            <version>${flink.version}version>
        dependency>
        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-clients_${scala.version}artifactId>
            <version>${flink.version}version>
        dependency>

        <dependency>
            <groupId>org.projectlombokgroupId>
            <artifactId>lombokartifactId>
            <version>1.16.22version>
        dependency>


        
        <dependency>
            <groupId>redis.clientsgroupId>
            <artifactId>jedisartifactId>
            <version>3.3.0version>
        dependency>

        
       <dependency>
           <groupId>com.alibabagroupId>
           <artifactId>fastjsonartifactId>
           <version>1.2.60version>
       dependency>


      
        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-table-api-java-bridge_${scala.version}artifactId>
            <version>${flink.version}version>
        dependency>
        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-table-planner-blink_${scala.version}artifactId>
            <version>${flink.version}version>
        dependency>
        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-table-commonartifactId>
            <version>${flink.version}version>
        dependency>

        
        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-cep-scala_${scala.version}artifactId>
            <version>${flink.version}version>
        dependency>

        
        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-csvartifactId>
            <version>${flink.version}version>
        dependency>

        
        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-connector-kafka_${scala.version}artifactId>
            <version>${flink.version}version>
        dependency>

        
        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-connector-filesystem_${scala.binary.version}artifactId>
            <version>1.4.2version>
        dependency>
        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-hadoop-compatibility_${scala.version}artifactId>
            <version>${flink.version}version>
        dependency>
        <dependency>
            <groupId>org.apache.hadoopgroupId>
            <artifactId>hadoop-clientartifactId>
            <version>${hadoop.version}version>
        dependency>
        <dependency>
            <groupId>org.apache.hadoopgroupId>
            <artifactId>hadoop-commonartifactId>
            <version>${hadoop.version}version>
        dependency>
        <dependency>
            <groupId>org.apache.hadoopgroupId>
            <artifactId>hadoop-hdfsartifactId>
            <version>${hadoop.version}version>
        dependency>

        
        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-jdbc_${scala.version}artifactId>
            <version>1.9.2version>
        dependency>
        <dependency>
            <groupId>mysqlgroupId>
            <artifactId>mysql-connector-javaartifactId>
            <version>5.1.38version>
        dependency>

        

        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-hbase_${scala.version}artifactId>
            <version>1.8.1version>
        dependency>

        <dependency>
            <groupId>org.apache.hbasegroupId>
            <artifactId>hbase-clientartifactId>
            <version>2.4.3version>
        dependency>

        
        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-connector-jdbc_${scala.version}artifactId>
            <version>1.13.0version>
        dependency>
        <dependency>
            <groupId>ru.yandex.clickhousegroupId>
            <artifactId>clickhouse-jdbcartifactId>
            <version>0.2.4version>
        dependency>
        
        <dependency>
            <groupId>com.google.guavagroupId>
            <artifactId>guavaartifactId>
            <version>30.1.1-jreversion>
        dependency>

        
         <dependency>
             <groupId>org.apache.flinkgroupId>
             <artifactId>flink-connector-jdbc_${scala.version}artifactId>
             <version>${flink.version}version>
         dependency>
         <dependency>
             <groupId>ru.yandex.clickhousegroupId>
             <artifactId>clickhouse-jdbcartifactId>
             <version>0.2.4version>
             <exclusions>
                 <exclusion>
                     <groupId>com.fasterxml.jackson.coregroupId>
                     <artifactId>jackson-databindartifactId>
                 exclusion>
                 <exclusion>
                     <groupId>com.fasterxml.jackson.coregroupId>
                     <artifactId>jackson-coreartifactId>
                 exclusion>
             exclusions>
         dependency>

        
         <dependency>
             <groupId>org.apache.bahirgroupId>
             <artifactId>flink-connector-redis_2.11artifactId>
             <version>1.0version>
         dependency>

        
        <dependency>
            <groupId>redis.clientsgroupId>
            <artifactId>jedisartifactId>
            <version>3.3.0version>
        dependency>

        
        <dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-connector-elasticsearch7_${scala.version}artifactId>
            <version>1.10.1version>
        dependency>

    dependencies>

project>

bean

 1 import lombok.AllArgsConstructor;
 2 import lombok.Data;
 3 import lombok.NoArgsConstructor;
 4 
 5 @Data
 6 @AllArgsConstructor
 7 @NoArgsConstructor
 8 public class Stu {
 9     private String name;
10     private String sub;
11     private Integer score;
12 }

test

 1 import org.apache.flink.api.common.functions.MapFunction;
 2 import org.apache.flink.api.java.tuple.Tuple2;
 3 import org.apache.flink.streaming.api.datastream.DataStream;
 4 import org.apache.flink.streaming.api.datastream.DataStreamSource;
 5 import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
 6 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 7 import org.apache.flink.table.api.*;
 8 import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
 9 import org.apache.flink.table.descriptors.*;
10 import org.apache.flink.table.descriptors.Schema;
11 import org.apache.flink.types.Row;
12 
13 
14 public class FlinkTEst {
15     public static void main(String[] args) throws Exception {
16         //创建TableEnvironment
17         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
18         env.setParallelism(1);
19         //流处理
20         EnvironmentSettings build = EnvironmentSettings.newInstance().inStreamingMode().useBlinkPlanner().build();
21         StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, build);
22         //批处理
23         EnvironmentSettings build2 = EnvironmentSettings.newInstance().inBatchMode().useBlinkPlanner().build();
24         TableEnvironment tabEnv = TableEnvironment.create(build2);
25         // 将以下数据注册为一个FTable。其中第一列是学生姓名，第二列是科目，第三列是分数 
26         //xiaoming english 90
27         //     xiaoming math 80
28         //     xiaohong english 98
29         //     xiaohong math 82
30         DataStreamSource source = env.readTextFile("src\\main\\resources\\file.txt");
31 
32         SingleOutputStreamOperator map = source.map(new MapFunction() {
33             @Override
34             public Stu map(String s) throws Exception {
35                 String[] ss = s.split(",");
36                 return new Stu(ss[0], ss[1],Integer.valueOf(ss[2]));
37             }
38         });
39         // 通过TABLE API查询所有数据并打印到控制台 
40         tableEnv.createTemporaryView("map",map);
41         Table table4 = tableEnv.sqlQuery("select * from map");
42         TableResult execute4 = table4.execute();
43         execute4.print();
44         // 统计学生总分数。 
45         Table table5 = tableEnv.sqlQuery("select name,sum(score) from map group by name");
46         TableResult execute5 = table5.execute();
47         execute5.print();
48         // 统计各科成绩的最高分 
49         Table table6 = tableEnv.sqlQuery("select sub,max(score) from map group by sub");
50         TableResult execute6 = table6.execute();
51         execute6.print();
52         // 查询出各科成绩的平均分 
53         Table table7 = tableEnv.sqlQuery("select sub,avg(score) from map group by sub");
54         TableResult execute7 = table7.execute();
55         execute7.print();
56         // 统计出总分最高的学生姓名 
57         tabEnv.connect(new FileSystem()
58                 .path("src\\main\\resources\\file.txt"))
59                 .withFormat(new Csv())
60                 .withSchema(new Schema()
61                         .field("name",DataTypes.STRING())
62                         .field("sub",DataTypes.STRING())
63                         .field("score",DataTypes.INT()))
64                 .createTemporaryTable("stu");
65 
66         Table table8 = tabEnv.sqlQuery("select name,sum(score) ss from stu group by name order by ss desc limit 1");
67         TableResult execute8 = table8.execute();
68         execute8.print();
69         //test
70         System.out.println("统计出总分最高的学生姓名");
71         Table table10 = tableEnv.sqlQuery("select name,sum(score) sums from map group by name order by sums desc limit 1");
72         TableResult execute10 = table10.execute();
73         execute10.print();
74 
75         System.out.println("Table toAppendStream test");
76      /*   DataStream rowDataStream = tableEnv.toAppendStream(table5, Row.class);
77         rowDataStream.print();*/
78 
79         System.out.println("Table toRetractStream test");
80         DataStream> tuple2DataStream = tableEnv.toRetractStream(table5, Row.class);
81         tuple2DataStream.print();
82 
83         //exe
84         env.execute();
85     }
86 }