flink使用DataStream API处理有界数据流

系统管理员 2022-10-15 12:48 160阅读 0赞

**目录**

主类

pom.xml

运行结果

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

## 主类 ##

package practice;
    
    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;
    
    /**
     * @author pangsl
     * @date 2021/5/25 9:08
     * @Description flink使用DataSet API处理有界数据流
     */
    public class BatchWordCount {
        public static void main(String[] args) throws Exception {
            //获取执行环境
            ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
            //加载或创建数据源
            DataSet<String> text = env.fromElements(
                    "Flink batch demo",
                    "batch demo",
                    "demo");
            //转换数据
            DataSet<Tuple2<String,Integer>> ds = text.flatMap(new LineSplitter()).groupBy(0).sum(1);
            //打印到控制台
            ds.print();
    
        }
    
        static class LineSplitter implements FlatMapFunction<String, Tuple2<String,Integer>> {
            //@Override
            public void flatMap(String line, Collector<Tuple2<String, Integer>> collector) throws Exception {
                //使用空格分割单词
                for (String word: line.split(" ")){
                    collector.collect(new Tuple2<String, Integer>(word,1));
                }
            }
        }
    }

## 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>
    
    
        <dependencies>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-core</artifactId>
                <version>1.12.0</version>
                <scope>provided</scope>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-java_2.11</artifactId>
                <version>1.12.0</version>
                <scope>provided</scope>
            </dependency>
    
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-clients_2.12</artifactId>
                <version>1.12.0</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/bc2dc0876d4a4545a9893902a0fe4eb3.png