【零基础学flink】flink实战：使用flink分析wiki log

「爱情、让人受尽委屈。」 2022-02-14 07:49 252阅读 0赞

本文将从头开始，从设置Flink项目到在Flink集群上运行流分析程序。 Wikipedia提供了一个IRC频道，其中记录了对Wiki的所有编辑。我们将在Flink中读取此通道，并计算每个用户在给定时间窗口内编辑的字节数。这很容易使用Flink在几分钟内实现，但它将为您提供一个良好的基础，从而开始自己构建更复杂的分析程序。

### 一、idea环境搭建 ###

使用idea新建maven项目，并把相关依赖包加入到pom.xml文件中，pom依赖如下：

<dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.11</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.11</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-wikiedits_2.11</artifactId>
            <version>${flink.version}</version>
        </dependency>
    </dependencies>

### 项目代码编写 ###

**tips**：在idea中自动生成main函数的快捷方法：输入psv后按`tab键`或者`enter`键即可自动生成main方法  
![1240][]

上面只是生成了main方法，万里长城刚刚开始。Flink程序的第一步是创建\*\*`StreamExecutionEnvironment`**（如果您正在编写batch处理作业，则创建`ExecutionEnvironment`）。它可用于**设置执行参数，并创建从外部系统读取的源\*\*。所以让我们继续把它添加到main方法：

StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment();

接下来我们将创建一个读取wikipedia 日志的source：

DataStream<WikipediaEditEvent> edits = see.addSource(new WikipediaEditsSource());

这创建了一个我们可以进一步处理的WikipediaEditEvent元素的DataStream。出于本示例的目的，我们感兴趣的是确定每个用户在特定时间窗口中添加或删除的字节数，比如说五秒。为此，我们首先要指定我们要在用户名上键入流，也就是说此流上的操作应考虑用户名。在我们的例子中，窗口中编辑的字节的总和应该是每个唯一的用户。为了键入一个Stream，我们必须提供一个KeySelector，如下所示：

DataStream<Tuple2<String, Long>> result = keyedEdits
        .timeWindow(Time.seconds(5))
        .fold(new Tuple2<>("", 0L), new FoldFunction<WikipediaEditEvent, Tuple2<String, Long>>() { 
            @Override
            public Tuple2<String, Long> fold(Tuple2<String, Long> acc, WikipediaEditEvent event) { 
                acc.f0 = event.getUser();
                acc.f1 += event.getByteDiff();
                return acc;
            }
        });

第一个调用.timeWindow（）指定我们希望有五秒钟的翻滚（非重叠）窗口。第二个调用为每个唯一键指定每个窗口切片的折叠变换。在我们的例子中，我们从初始值（“”，0L）开始，并在该时间窗口中为用户添加每个编辑的字节差异。生成的Stream现在包含每隔五秒发出一次的每个用户的Tuple2 <String，Long>。 剩下要做的就是将流打印到控制台并开始执行：

result.print();
    
    see.execute();

最后一次调用是启动实际Flink工作所必需的。所有操作（例如创建源，转换和接收器）仅构建内部操作的图形。只有在调用execute（）时，才会在集群上抛出或在本地计算机上执行此操作图。

### 运行结果 ###

1> (Fenix down,114)
    6> (AnomieBOT,155)
    8> (BD2412bot,-3690)
    7> (IgnorantArmies,49)
    3> (Ckh3111,69)
    5> (Slade360,0)
    7> (Narutolovehinata5,2195)
    6> (Vuyisa2001,79)
    4> (Ms Sarah Welch,269)
    4> (KasparBot,-245)

完整源代码：

import org.apache.flink.api.common.functions.FoldFunction;
    import org.apache.flink.api.java.functions.KeySelector;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.KeyedStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.windowing.time.Time;
    import org.apache.flink.streaming.connectors.wikiedits.WikipediaEditEvent;
    import org.apache.flink.streaming.connectors.wikiedits.WikipediaEditsSource;
    
    public class wikiedits { 
        public static void main(String[] args) throws Exception{ 
            StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment();
            DataStream<WikipediaEditEvent> edits = see.addSource(new WikipediaEditsSource());
    
            KeyedStream<WikipediaEditEvent, String> keyedEdits = edits
                    .keyBy(new KeySelector<WikipediaEditEvent, String>() { 
                        public String getKey(WikipediaEditEvent event) { 
                            return event.getUser();  //按照用户名组织数据
                        }
                    });
            DataStream<Tuple2<String, Long>> result = keyedEdits
                    .timeWindow(Time.seconds(5))
                    .fold(new Tuple2<String, Long>("", 0L), new FoldFunction<WikipediaEditEvent, Tuple2<String, Long>>() { 
                        public Tuple2<String, Long> fold(Tuple2<String, Long> acc, WikipediaEditEvent event) { 
                            acc.f0 = event.getUser();
                            acc.f1 += event.getByteDiff();
                            return acc;
                        }
                    });
            result.print();  // 也可以将result写入文件或者kafka
    
            see.execute();
        }
    }

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

扫描下方二维码，**及时**获取更多**互联网求职面经**、**java**、**python**、**爬虫**、**大数据**等技术，和**海量资料分享**：  
公众号\*\*`菜鸟名企梦`**后台发送“csdn”即可免费领取【csdn】和【百度文库】下载服务；  
公众号**`菜鸟名企梦`**后台发送“资料”:即可领取**5T精品学习资料\*\*、**java面试考点**和**java面经总结**，以及**几十个java、大数据项目**，**资料很全，你想找的几乎都有**  
![扫码关注，及时获取更多精彩内容。（博主今日头条大数据工程师）][1240 1]

[1240]: /images/20220214/997c98d3caab46538aa96eea94526f65.png
[1240 1]: /images/20220214/8ef05557de234b499ba8bfb82892bbfb.png