Flink练习第一天:wordcount、批处理、流处理

Flink练习第一天:wordcount、批处理、流处理,第1张

目录

新建工程,引入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();


    }

}
运行结果:

欢迎分享,转载请注明来源:内存溢出

原文地址: https://outofmemory.cn/langs/868382.html

(0)
打赏 微信扫一扫 微信扫一扫 支付宝扫一扫 支付宝扫一扫
上一篇 2022-05-13
下一篇 2022-05-13

发表评论

登录后才能评论

评论列表(0条)

保存