flink使用SQL处理无界数据流

浅浅的花香味﹌ 2022-10-15 13:53 160阅读 0赞

**目录**

主类

MySource类

pom.xml

结果输出

--------------------

## 主类 ##

package myflink.sql;
    
    import myflink.source.MySource;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.table.api.Table;
    import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
    import org.apache.flink.types.Row;
    
    import static org.apache.flink.table.api.Expressions.$;
    
    /**
     * @author pangsl
     * @date 2021/5/25 15:13
     * @Description flink使用SQL处理无界数据流
     */
    public class SQLStreamDemo {
    
        public static void main(String[] args) throws Exception{
    
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
    
            DataStream<String> stream= env.addSource(new MySource());
    
            Table table = tEnv.fromDataStream(stream, $("word"));
    
            Table result = tEnv.sqlQuery("select * from " + table + " where word like '%t%'");
    
            tEnv.toAppendStream(result, Row.class)
                    .print();
    
            env.execute();
    
        }
    }

## MySource类 ##

package myflink.source;
    
    import org.apache.flink.streaming.api.functions.source.SourceFunction;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.Random;
    
    /**
     * @author pangsl
     * @date 2021/5/25 10:19
     * @Description
     */
    public class MySource implements SourceFunction<String> {
        private long count = 1L;
        private boolean isRunning = true;
    
        //通过循环产生数据
        public void run(SourceContext<String> ctx) throws Exception {
            while (isRunning){
                //单词流
                List<String> stringList = new ArrayList<String>();
                stringList.add("world");
                stringList.add("flink");
                stringList.add("stream");
                stringList.add("batch");
                stringList.add("table");
                stringList.add("sql");
                stringList.add("hello");
    
                int size = stringList.size();
                int i = new Random().nextInt(size);
                ctx.collect(stringList.get(i));
                //每秒产生一条数据
                Thread.sleep(1000);
    
            }
        }
        //取消执行
        public void cancel() {
            isRunning = false;
        }
    }

## pom.xml ##

<?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">
        <parent>
            <artifactId>flinkdemo</artifactId>
            <groupId>com.flink.pangsl</groupId>
            <version>1.0-SNAPSHOT</version>
        </parent>
        <modelVersion>4.0.0</modelVersion>
    
        <artifactId>Flink_shizhanpai</artifactId>
        <build>
            <plugins>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <configuration>
                        <source>8</source>
                        <target>8</target>
                    </configuration>
                </plugin>
            </plugins>
        </build>
    
    
        <properties>
            <flink.version>1.11.0</flink.version>
            <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        </properties>
    
        <dependencies>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-core</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-java_2.11</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-clients_2.11</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-table-api-scala-bridge_2.11</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-table-planner_2.11</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-table-planner-blink_2.11</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-scala_2.11</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
    
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-table-common</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
    
    
    
        </dependencies>
    
    </project>

## 结果输出 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTExMTAzMDE_size_16_color_FFFFFF_t_70][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTExMTAzMDE_size_16_color_FFFFFF_t_70]: /images/20221014/70ee518f07004ec4b1e735f9498bf5e3.png