flink实例开发-详细使用指南

Flink入门及实战-上：

http://edu.51cto.com/sd/07245

Flink入门及实战-下：

http://edu.51cto.com/sd/5845e

flink实例开发-详细使用指南

 

• 配置一个maven项目
• 编写一个flink程序
• 编程实战：编写一个向kafka写数据的程序在集群运行 flink整合kafka

在本指南中，我们将从头开始，从flink项目设置到在集群上运行一个实时流程序。

维基百科提供了一个IRC管道记录了所有用户的编辑记录日志，我们可以通过flink读取这个管道的内容在一个时间窗口内进行计数操作，这是很简单的，使用flink几分钟就可以实现，但是它会给你一个良好的基础，开始构建自己更复杂的分析程序。

 

配置一个maven项目

我们将使用flink的maven archetype模板创建一个项目架构，请点击此链接Java API Quickstart查看更多详细信息。

执行下面的命令创建项目：

$ mvn archetype:generate \
    -DarchetypeGroupId=org.apache.flink \
    -DarchetypeArtifactId=flink-quickstart-java \
    -DarchetypeVersion=1.4.1 \
    -DgroupId=wiki-edits \
    -DartifactId=wiki-edits \
    -Dversion=0.1 \
    -Dpackage=wikiedits \
    -DinteractiveMode=false

你可以修改 groupId artifactId 和 package的名称。使用上面的参数，maven将会创建一个类似下面的项目架构。

$ tree wiki-edits
wiki-edits/
├── pom.xml
└── src
    └── main
        ├── java
        │   └── wikiedits
        │       ├── BatchJob.java
        │       ├── SocketTextStreamWordCount.java
        │       ├── StreamingJob.java
        │       └── WordCount.java
        └── resources
            └── log4j.properties

这里面的pom.xml文件已经添加了flink的相关依赖和几个实例项目实例。flink程序在 src/main/java 目录下面。我们可以删除这写示例程序，可以执行下面的命令删除：

rm wiki-edits/src/main/java/wikiedits/*.java

最后一步我们需要添加维基百科connector作为依赖，为了我们可以在程序中使用它。修改pom.xml的dependencies依赖，看起来像下面这样：

注意：下面配置中的${flink.version}是一个变量，代表的是flink的版本号，如果pom.xml中没有定义这个变量的话，建议修改为实际的flink版本号，例如：<version>1.4.1</version>

<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>

注意检查一下 flink-connector-wikiedits_2.11 依赖是否添加进来。

 

编写一个flink程序

下面是编码时间。启动你最喜欢的IDE工具(idea或者eclipse)，并且导入刚才创建的maven项目，在对应的目录下创建一个文件：src/main/java/wikiedits/WikipediaAnalysis.java

package wikiedits;

public class WikipediaAnalysis {

    public static void main(String[] args) throws Exception {

    }
}

这个代码现在是非常基本的，我们将会去填满它。请注意，在这里我不会把idea可以自动填充的import语句展示出来。在本节的最后，我将展示完整的代码和import导入语句。

flink程序的第一步是创建一个StreamExecutionEnvironment (或者是ExecutionEnvironment，如果你编写的是一个batch批处理job)。这可以用于设置执行参数和可以设置一个输入源来从外部系统读取数据。让我们继续添加这个main方法

StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment();

下一步我将会创建一个source输入源，来读取维基百科的IRC日志数据。

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

现在创建了一个 wikipediaEditEvent的DataStream数据流，然后我们可以进一步对数据进行处理。对于本示例中，我们感兴趣的是每个用户在特性时间窗口内的添加或者删除字节数量，假设是5秒钟。我们首先必须指定，我们想获取数据流中的用户名，也就是说操作这个流程应该考虑用户名。在我们的例子中，应该统计的是在指定时间窗口中每个用户的编辑字节量的和。如果想对流进行key操作，我们必须提供一个keySelector，像下面这样：

KeyedStream<WikipediaEditEvent, String> keyedEdits = edits
    .keyBy(new KeySelector<WikipediaEditEvent, String>() {
        @Override
        public String getKey(WikipediaEditEvent event) {
            return event.getUser();
        }
    });

这个方法返回给我们了一个wikipediaEditEvent的数据流，并且包含了一个string类型的key，用户名。我们现在可以在这个数据流上指定一个时间窗口，并且可以基于这个时间窗口计算元素的结果。一个窗口指定一个流片段进行计算。当在一个无边界的流中进行聚合计算的时候，必须要指定一个window时间窗口。在我们的例子中，我们会说我们想每5秒钟对用户编辑的字节数量进行求和。

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()，指定了我们想要统计的滚动时间窗口(不重叠的)是5秒钟。第二步调用了fold转换算子在每个时间窗口为每个唯一的key。在我们的例子中，我们从一个初始值开始("", 0L)，并且针对每一个用户在指定的时间窗口内添加对应的字节变化值。最终生成的结束数据流包含了一个tuple2<String,Long>，这个tuple2中的数据是指每5秒钟针对每个用户统计的字节变化量。

剩下要做的唯一的事情就是把流打印到控制台，并且开始执行。

result.print();
//调用execute方法 才会真正开始执行
see.execute();

最后一个方法调用是启动flink job的必要条件。所有操作，例如创建source数据源，转换，和sink操作仅仅是建立一个图的内部操作。只有当执行execute方法的时候，这个图的所有操作才会在本机或者放到集群运行。

 

到目前为止完整代码如下：

package wikiedits;

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 WikipediaAnalysis {

  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>() {
        @Override
        public String getKey(WikipediaEditEvent event) {
          return event.getUser();
        }
      });

    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;
        }
      });

    result.print();

    see.execute();
  }
}

你可以运行这个例子在ide中或者在命令行中，使用maven：

$ mvn clean package
$ mvn exec:java -Dexec.mainClass=wikiedits.WikipediaAnalysis

上面的第一个命令是对项目进行编译打包，第二个命令是执行我们的入口类。输出内容类似于下面这样：

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)

每一行前面的数字代表这一行是哪一个并行线程输出的。

 

编程实战：编写一个向kafka写数据的程序在集群运行

在做这个实例之前，请先安装好flink集群和kafka集群。

第一步，我们必须在pom.xml文件中添加kafka依赖：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka-0.8_2.11</artifactId>
    <version>${flink.version}</version>
</dependency>

下一步，我们需要修改代码，移除掉print() 这个sink组件，然后使用kafka sink进行替代，新代码如下：

result
    .map(new MapFunction<Tuple2<String,Long>, String>() {
        @Override
        public String map(Tuple2<String, Long> tuple) {
            return tuple.toString();
        }
    })
    .addSink(new FlinkKafkaProducer08<>("localhost:9092", "wiki-result", new SimpleStringSchema()));

相关的类也需要导入：

import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer08;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.functions.MapFunction;

注意我们第一次使用mapFunction把tuple2<String,Long>流转换为一个string类型的流，我们这样做的目的是为了很方便的吧数据写入kafka。然后，我们创建建了一个kafka sink。你可能需要调整代码中的kafka主机名和端口号。"wiki-result"是我们在运行程序之前，要创建的kafka topic名称。下面使用maven命令构建项目，获取构建的jar包，需要先进入到项目根目录下面，然后再执行下面命令。

$ mvn clean package

编译生成的jar包会在target目录下面：target/wiki-edits-0.1.jar 后面我会用到这个jar包

现在我们需要准备好flink集群为了启动程序往kafka中写数据。进入到你安装flink的位置启动flin集群：

$ cd my/flink/directory
$ bin/start-local.sh

我们也需要创建kafka tpic，以便于我们的程序可以向里面写数据

$ cd my/kafka/directory
$ bin/kafka-topics.sh --create --zookeeper localhost:2181 --topic wiki-result

现在我们准备在本地集群上运行我们的jar包

$ cd my/flink/directory
$ bin/flink run -c wikiedits.WikipediaAnalysis path/to/wikiedits-0.1.jar

如果一切运行正常，你的输出应该是类似这样的：

03/08/2016 15:09:27 Job execution switched to status RUNNING.
03/08/2016 15:09:27 Source: Custom Source(1/1) switched to SCHEDULED
03/08/2016 15:09:27 Source: Custom Source(1/1) switched to DEPLOYING
03/08/2016 15:09:27 TriggerWindow(TumblingProcessingTimeWindows(5000), FoldingStateDescriptor{name=window-contents, defaultValue=(,0), serializer=null}, ProcessingTimeTrigger(), WindowedStream.fold(WindowedStream.java:207)) -> Map -> Sink: Unnamed(1/1) switched to SCHEDULED
03/08/2016 15:09:27 TriggerWindow(TumblingProcessingTimeWindows(5000), FoldingStateDescriptor{name=window-contents, defaultValue=(,0), serializer=null}, ProcessingTimeTrigger(), WindowedStream.fold(WindowedStream.java:207)) -> Map -> Sink: Unnamed(1/1) switched to DEPLOYING
03/08/2016 15:09:27 TriggerWindow(TumblingProcessingTimeWindows(5000), FoldingStateDescriptor{name=window-contents, defaultValue=(,0), serializer=null}, ProcessingTimeTrigger(), WindowedStream.fold(WindowedStream.java:207)) -> Map -> Sink: Unnamed(1/1) switched to RUNNING
03/08/2016 15:09:27 Source: Custom Source(1/1) switched to RUNNING

你将会看到各个组件开始运行，这里只有两个，因为窗口之后的操作并入成了一个。在flink中我们称之为 chaining

你可以起到一个kafka的控制台消费者来观察程序的输出：

bin/kafka-console-consumer.sh  --zookeeper localhost:2181 --topic wiki-result

你也可以查看flink的dashboard界面，http://localhost:8081 可以在这里面看到集群资源的情况和任务的执行状况。

如果你点击进入你启动的job中你将会得到一个视图，并且你可以检查单个操作，例如：看到处理元素的数量：

 

 

 

获取更多大数据资料，视频以及技术交流请加群：