使用Maven构建Flink第一个项目

公众号guangcity

发布于 2019-09-20 17:26:10

6K0

发布于 2019-09-20 17:26:10

文章被收录于专栏：光城(guangcity)光城(guangcity)

使用Maven构建Flink第一个项目

0.准备工作

下载Flink并运行！

1.IDEA创建Maven项目

点击左上角File->new Project->后面直接点击下一步->GroupID与artifactId、version随便填写->填写项目名称，点击完成。

注意：Maven打包后的jar包名字就是我们artifactId-version.jar格式，可以适当根据喜好在pom.xml修改。

pom.xml修改：

<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_2.11</artifactId> <!--2.11为scala.binary.version-->
        <version>1.7.2</version> <!--Flink版本号-->
    </dependency>
</dependencies>

点击import change即可。

新java class命名为：SocketTextStreamWordCount。

填写下面代码：

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class SocketTextStreamWordCount {
    public static void main(String[] args) throws Exception {
        //参数检查
        if (args.length != 2) {
            System.err.println("USAGE:\nSocketTextStreamWordCount <hostname> <port>");
            return;
        }

        String hostname = args[0];
        Integer port = Integer.parseInt(args[1]);


        // set up the streaming execution environment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //获取数据
        DataStreamSource<String> stream = env.socketTextStream(hostname, port);

        //计数
        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = stream.flatMap(new LineSplitter())
                .keyBy(0)
                .sum(1);

        sum.print();

        env.execute("Java WordCount from SocketTextStream Example");
    }

    public static final class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> {

        public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) {
            String[] tokens = s.toLowerCase().split("\\W+");

            for (String token: tokens) {
                if (token.length() > 0) {
                    collector.collect(new Tuple2<String, Integer>(token, 1));
                }
            }
        }
    }
}