目录

新建工程，引入pom.xml

批处理代码

 文件准备：在工程下新建文件夹input，新建文件wordcount.txt

运行结果：

有界流处理代码

 运行结果：

 无界流处理代码（需要开启cocket连接）

运行结果：

新建工程，引入pom.xml

    4.0.0

    com.flink
    Flink_Demo
    1.0-SNAPSHOT

    
        1.13.0
        1.8
        2.12
        1.7.30
    


    
        
            org.apache.flink
            flink-java
            ${flink.version}
        
        
            org.apache.flink
            flink-streaming-java_${scala.binary.version}
            ${flink.version}
        
        
            org.apache.flink
            flink-clients_${scala.binary.version}
            ${flink.version}
        
        
            org.slf4g
            slf4g-api
            ${slf4j.version}
        
        
            org.slf4g
            slf4g-log4j12
            ${slf4j.version}
        
        
            org.apache.logging.log4j
            log4j-to-slf4j
            2.14.0
        


    


    
        
            
                org.apache.maven.plugins
                maven-assembly-plugin
                3.0.0
                
                    
                        jar-with-dependencies
                    
                
                
                    
                        make-assembly
                        package
                        
                            single
                        
                    
                

            
        
    

批处理代码

package com.atguigu.wc;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;



public class WordCount {
    public static void main(String[] args) throws Exception{

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSet stringDataSource = env.readTextFile("D:\develop\idea_workspace\Flink_Demo_SGG\input\wordcount.txt");
        DataSet> resultSet = stringDataSource.flatMap(new MyFlatMapper()).groupBy(0).sum(1);
        resultSet.print();
    }

    static class MyFlatMapper implements FlatMapFunction> {
        public void flatMap(String value, Collector> out) throws Exception {
            String[] words = value.split(" ");
            //words遍历
            for (String word:words){
                out.collect(new Tuple2(word, (long) 1));
            }

        }

    }
}

 文件准备：在工程下新建文件夹input，新建文件wordcount.txt

hello word
python java flink
hello python java
java

运行结果：

有界流处理代码

package com.atguigu.wc;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class WordCount_Stream {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream stringDataSource = env.readTextFile("D:\develop\idea_workspace\Flink_Demo_SGG\input\wordcount.txt");

        SingleOutputStreamOperator> resultStream = stringDataSource.flatMap(new WordCount.MyFlatMapper()).keyBy(0).sum(1);
        resultStream.print();
        env.execute();
    }
}

 运行结果：

 无界流处理代码（需要开启cocket连接）

package com.atguigu.wc;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class WordCount_Stream_socket {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        DataStreamSource dataStreamSource = env.socketTextStream("10.22.82.120", 7777);
        DataStreamSource dataStreamSource = env.socketTextStream("10.30.202.21", 7777);

        SingleOutputStreamOperator> returns = dataStreamSource.flatMap(new WordCount.MyFlatMapper()).returns(Types.TUPLE(Types.STRING, Types.LONG));


        KeyedStream, String> stream = returns.keyBy(data -> data.f0);


        SingleOutputStreamOperator> sum = stream.sum(1);
        sum.print();

        env.execute();


    }

}

运行结果：