Flink的有界流和无界流

悠悠 2022-11-04 08:03 194阅读 0赞

# Flink的有界流和无界流 #

## 一、创建maven项目 ##

POM文件中添加需要的依赖:

<properties>
        <flink.version>1.12.0</flink.version>
        <java.version>1.8</java.version>
        <scala.binary.version>2.11</scala.binary.version>
        <slf4j.version>1.7.30</slf4j.version>
    </properties>
    
    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
    	   <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
    	   <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
    	   <scope>provided</scope>
        </dependency>
    
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-runtime-web_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
    
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>${slf4j.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>${slf4j.version}</version>
    	   <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-to-slf4j</artifactId>
            <version>2.14.0</version>
        </dependency>
    </dependencies>
    
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>3.3.0</version>
                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

src/main/resources添加文件:log4j.properties

log4j.rootLogger=error, stdout
    log4j.appender.stdout=org.apache.log4j.ConsoleAppender
    log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
    log4j.appender.stdout.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n

配置idea, 运行的时候包括provided scope  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]

## 二、批处理WordCount ##

// 1. 创建执行环境
    ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    // 2. 从文件读取数据 按行读取(存储的元素就是每行的文本)
    DataSource<String> lineDS = env.readTextFile("input/words.txt");
    // 3. 转换数据格式
    FlatMapOperator<String, Tuple2<String, Long>> wordAndOne = lineDS
            .flatMap((String line, Collector<Tuple2<String, Long>> out) -> { 
                String[] split = line.split(" ");
                for (String word : split) { 
                    out.collect(Tuple2.of(word, 1L));
                }
            })
            .returns(Types.TUPLE(Types.STRING, Types.LONG));  // 当Lambda表达式使用 java 泛型的时候, 由于泛型擦除的存在, 需要显示的声明类型信息
    
    // 4. 按照 word 进行分组
    UnsortedGrouping<Tuple2<String, Long>> wordAndOneUG = wordAndOne.groupBy(0);
    // 5. 分组内聚合统计
    AggregateOperator<Tuple2<String, Long>> sum = wordAndOneUG.sum(1);
    
    // 6. 打印结果
    sum.print();

## 三、流处理WordCount ##

### 3.1有界流 ###

// 1. 创建流式执行环境
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    // 2. 读取文件
    DataStreamSource<String> lineDSS = env.readTextFile("input/words.txt");
    // 3. 转换数据格式
    SingleOutputStreamOperator<Tuple2<String, Long>> wordAndOne = lineDSS
            .flatMap((String line, Collector<String> words) -> { 
                Arrays.stream(line.split(" ")).forEach(words::collect);
            })
            .returns(Types.STRING)
            .map(word -> Tuple2.of(word, 1L))
            .returns(Types.TUPLE(Types.STRING, Types.LONG));
    // 4. 分组
    KeyedStream<Tuple2<String, Long>, String> wordAndOneKS = wordAndOne
            .keyBy(t -> t.f0);
    // 5. 求和
    SingleOutputStreamOperator<Tuple2<String, Long>> result = wordAndOneKS
            .sum(1);
    // 6. 打印
    result.print();
    // 7. 执行
    env.execute();

### 3.2无界流 ###

// 1. 创建流式执行环境
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    // 2. 读取文件
    DataStreamSource<String> lineDSS = env.socketTextStream("hadoop162", 9999);
    // 3. 转换数据格式
    SingleOutputStreamOperator<Tuple2<String, Long>> wordAndOne = lineDSS
            .flatMap((String line, Collector<String> words) -> { 
                Arrays.stream(line.split(" ")).forEach(words::collect);
            })
            .returns(Types.STRING)
            .map(word -> Tuple2.of(word, 1L))
            .returns(Types.TUPLE(Types.STRING, Types.LONG));
    // 4. 分组
    KeyedStream<Tuple2<String, Long>, String> wordAndOneKS = wordAndOne
            .keyBy(t -> t.f0);
    // 5. 求和
    SingleOutputStreamOperator<Tuple2<String, Long>> result = wordAndOneKS
            .sum(1);
    // 6. 打印
    result.print();
    // 7. 执行
    env.execute();

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]: /images/20221023/611cdeba0ab148d5a62c0305a73ee909.png