一文入门流处理开发

数据社

发布于 2020-05-21 11:38:30

5350

发布于 2020-05-21 11:38:30

文章被收录于专栏：数据社数据社

一、Flink介绍

Apache Flink 是一个开源的分布式流处理和批处理系统。Flink 的核心是在数据流上提供数据分发、通信、具备容错的分布式计算。同时，Flink 在流处理引擎上构建了批处理引擎，原生支持了迭代计算、内存管理和程序优化。

详细介绍可以参考之前的《flink简介》。

二、部署环境

操作系统环境：

flink支持Linux, Mac OS X, 和 Windows环境部署，本次部署选择Linux环境部署。

JDK：要求Java 7或者更高

三、下载软件

jdk1.8.0_144
flink-1.4.2-bin-hadoop26-scala_2.11.tgz

四、部署步骤

1、JDK安装步骤此处省略，安装后验证下JDK环境

$ java -version

openjdk version "1.8.0_144"

OpenJDKRuntimeEnvironment(build 1.8.0_144-b01)

OpenJDK64-BitServer VM (build 25.144-b01, mixed mode)

2、安装部署flink 本文介绍flink部署分为两种模式：local，standalone。下面依次介绍这两种模式的部署方式。

找到下载的flink压缩包，进行解压

$ tar -zxvf flink-1.4.2-bin-hadoop26-scala_2.11.tgz

首先是local模式，最为简单。

$ cd flink-1.4.2

$ bin/start-local.sh

Starting job manager

我们可以通过查看日志确认是否启动成功

$ tailf flink-csap-taskmanager-0-XXXX.log

2018-05-0310:07:53,718 INFO  org.apache.flink.runtime.filecache.FileCache- User file cache uses directory /tmp/flink-dist-cache-4c371de9-0f85-4889-b4d9-4a522641549c

2018-05-0310:07:53,725 INFO  org.apache.flink.runtime.taskmanager.TaskManager- StartingTaskManager actor at akka://flink/user/taskmanager#-524742300.

2018-05-0310:07:53,725 INFO  org.apache.flink.runtime.taskmanager.TaskManager- TaskManager data connection information: 2c358d6f38949f9aae31c5bddb0cc1dc@ LY1F-R021707-VM14.local(dataPort=55234)

2018-05-0310:07:53,726 INFO  org.apache.flink.runtime.taskmanager.TaskManager- TaskManager has 1 task slot(s).

2018-05-0310:07:53,727 INFO  org.apache.flink.runtime.taskmanager.TaskManager- Memory usage stats: [HEAP: 111/1024/1024 MB, NON HEAP: 35/36/-1 MB (used/committed/max)]

2018-05-0310:07:53,730 INFO  org.apache.flink.runtime.taskmanager.TaskManager- Trying to register at JobManager akka.tcp://flink@localhost:6123/user/jobmanager (attempt 1, timeout: 500 milliseconds)

2018-05-0310:07:53,848 INFO  org.apache.flink.runtime.taskmanager.TaskManager- Successful registration at JobManager(akka.tcp://flink@localhost:6123/user/jobmanager), starting network stack and library cache.

2018-05-0310:07:53,851 INFO  org.apache.flink.runtime.taskmanager.TaskManager- Determined BLOB server address to be localhost/127.0.0.1:52382.Starting BLOB cache.

2018-05-0310:07:53,858 INFO  org.apache.flink.runtime.blob.PermanentBlobCache- Created BLOB cache storage directory /tmp/blobStore-c07b9e80-41f0-490f-8126-7008144c4b0b

2018-05-0310:07:53,861 INFO  org.apache.flink.runtime.blob.TransientBlobCache- Created BLOB cache storage directory /tmp/blobStore-e0d1b687-1c47-41c4-b5bc-10ceaa39e778

JobManager进程将会在8081端口上启动一个WEB页面，我们可以通过浏览器到hostname:8081中查看相关的信息。可以打开页面查看到相关信息，说明local模式部署是没问题的。

下面来看一下standlone部署方式。

安装JDK，解压压缩包，都是一样的。不一样的是我们要修改解压后的flink配置文件。然后在集群主机间做免密，免密操作方法。

修改conf/flink-conf.yaml，我们将jobmanager.rpc.address的值设置成你master节点的IP地址。此外，我们通过jobmanager.heap.mb和taskmanager.heap.mb配置参数来设置每个节点的JVM能够分配的最大内存。从配置参数名字可以看出，这个参数的单位是MB，如果某些节点拥有比你之前设置的值更多的内存时，我们可以在那个节通过FLINKTMHEAP参数类覆盖值钱的设置。

我们需要把所有将要作为worker节点的IP地址存放在conf/slaves文件中，在conf/slaves文件中，每个IP地址必须放在一行，如下：

192.168.0.100

192.168.0.101

.

.

.

192.168.0.150

然后将修改好的flink包整理复制到集群各个节点。每个节点flink路径保持一致。然后启动集群

$ bin/start-cluster.sh

查看日志是否成功。

以上是部署方法，部署成功后，我们来跑一个demo程序，验证一下Flink的流处理功能，对其有个初步的了解。

flink为了更好的让大家理解，已经给大家提供了一些demo代码，demo的jar包可以在/examples/streaming首先看一下demo代码：

objectSocketWindowWordCount{


def main(args: Array[String]) : Unit= {


// the port to connect to

        val port: Int= try{

ParameterTool.fromArgs(args).getInt("port")

} catch{

case e: Exception=> {

System.err.println("No port specified. Please run 'SocketWindowWordCount --port <port>'")

return

}

}


// get the execution environment

        val env: StreamExecutionEnvironment= StreamExecutionEnvironment.getExecutionEnvironment


// get input data by connecting to the socket

        val text = env.socketTextStream("localhost", port, '\n')


// parse the data, group it, window it, and aggregate the counts

        val windowCounts = text

.flatMap { w => w.split("\\s") }

.map { w => WordWithCount(w, 1) }

.keyBy("word")

.timeWindow(Time.seconds(5), Time.seconds(1))

.sum("count")


// print the results with a single thread, rather than in parallel

        windowCounts.print().setParallelism(1)


        env.execute("Socket Window WordCount")

}


// Data type for words with count

caseclassWordWithCount(word: String, count: Long)

}

这个demo是监控端口，然后对端口输入单子进行wordcount的程序。

运行demo，首先打开一个窗口进行端口数据输入：

$ nc -l 9001

hello

hello

word

world

然后运行demo监控端口单词输入统计：

$ ./bin/flink run examples/streaming/SocketWindowWordCount.jar --port 9001

运行后可以看到结果统计：

$ more flink-csap-taskmanager-0-XXX.out.1

hello : 1

hello : 1

word : 1

world : 1

五、IEDA开发环境搭建

1、安装java环境

~~此处略去，这个你已经会了~~~

2、安装maven

参考Maven安装与配置（https://www.jianshu.com/p/183d4f229103）

3、配置IDEA

参考如何使用IntelliJ IDEA 配置Maven（https://www.jianshu.com/p/467bd544998d）

4、pom文件设置

    </properties>

    <dependencies>

    <dependency>

    <groupId>org.scala-lang</groupId>

    <artifactId>scala-library</artifactId>

    <version>${scala.version}</version>

    </dependency>

    <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_${scala.binary.version}</artifactId>

    <version>${flink.version}</version>

    </dependency>

    <dependency>

    <groupId>org.apache.flink</groupId>

    <artifactId>flink-scala_${scala.binary.version}</artifactId>

    <version>${flink.version}</version>

    </dependency>

    <dependency>

    <groupId>org.apache.flink</groupId>

    <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>

    <version>${flink.version}</version>

    </dependency>

    <dependency>

    <groupId>org.apache.flink</groupId>

    <artifactId>flink-table_${scala.binary.version}</artifactId>

    <version>${flink.version}</version>

    </dependency>

    <dependency>

    <groupId>org.apache.flink</groupId>

    <artifactId>flink-clients_${scala.binary.version}</artifactId>

    <version>${flink.version}</version>

    </dependency>

    <dependency>

    <groupId>org.apache.flink</groupId>

    <artifactId>flink-connector-kafka-0.10_${scala.binary.version}</artifactId>

    <version>${flink.version}</version>

    </dependency>

    <dependency>

5、代码示例

import org.apache.flink.api.common.functions.FlatMapFunction;

import org.apache.flink.api.java.utils.ParameterTool;

import org.apache.flink.streaming.api.datastream.DataStream;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.windowing.time.Time;

import org.apache.flink.util.Collector;


/**

 * Author: qincf

 * Date: 2018/11/02

 * Desc: 使用flink对指定窗口内的数据进行实时统计，最终把结果打印出来

 *       先在目标主机1.1.1.1机器上执行nc -l 9000

 */

publicclassStreamingWindowWordCount{

publicstaticvoid main(String[] args) throwsException{

//定义socket的端口号

int port;

try{

ParameterTool parameterTool = ParameterTool.fromArgs(args);

            port = parameterTool.getInt("port");

}catch(Exception e){

System.err.println("没有指定port参数，使用默认值9000");

            port = 9000;

}

//获取运行环境

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

//连接socket获取输入的数据

DataStreamSource<String> text = env.socketTextStream("1.1.1.1", port, "\n");

//计算数据

DataStream<WordWithCount> windowCount = text.flatMap(newFlatMapFunction<String, WordWithCount>() {

publicvoid flatMap(String value, Collector<WordWithCount> out) throwsException{

String[] splits = value.split("\\s");

for(String word:splits) {

out.collect(newWordWithCount(word,1L));

}

}

})//打平操作，把每行的单词转为<word,count>类型的数据

//针对相同的word数据进行分组

.keyBy("word")

//指定计算数据的窗口大小和滑动窗口大小

.timeWindow(Time.seconds(2),Time.seconds(1))

.sum("count");

//获取可视化JSON

System.out.println(env.getExecutionPlan());

//把数据打印到控制台,使用一个并行度

        windowCount.print().setParallelism(1);

//注意：因为flink是懒加载的，所以必须调用execute方法，上面的代码才会执行

        env.execute("streaming word count");




}


/**

     * 主要为了存储单词以及单词出现的次数

     */

publicstaticclassWordWithCount{

publicString word;

publiclong count;

publicWordWithCount(){}

publicWordWithCount(String word, long count) {

this.word = word;

this.count = count;

}


@Override

publicString toString() {

return"WordWithCount{"+

"word='"+ word + '\''+

", count="+ count +

'}';

}

}


}

6、测试步骤

首先在1.1.1.1机器上使用nc命令模拟数据发送

nc -l 9000

然后在IEDA中运营StreamingWindowWordCount程序在主机上输入字符

[root@data01]# nc -l 9000

a

a

b

c

d

d

此时运行程序后，IDEA中会打印处结果

    WordWithCount{word='a', count=1}

    WordWithCount{word='a', count=2}

    WordWithCount{word='b', count=1}

    WordWithCount{word='d', count=1}

    WordWithCount{word='c', count=1}

    WordWithCount{word='c', count=1}

    WordWithCount{word='a', count=1}

    WordWithCount{word='d', count=1}

    WordWithCount{word='b', count=1}

大家会看到，wordcount的结果。仔细看还有一串json输出，这部分是什么呢？代码中加了一个打印执行计划的部分：

/获取可视化JSON

System.out.println(env.getExecutionPlan());

Flink提供了一个可视化执行计划的结果，类似Spark的DAG图，把json粘贴到Flink Plan Visualizer可以看到执行计划图：

现在你已经搭建好Flink开发环境了，可以开启你的流处理旅程了，更多教程可以参考Flink官网。

本文参与腾讯云自媒体分享计划，分享自微信公众号。

原始发表：2020-05-09，如有侵权请联系 cloudcommunity@tencent.com 删除

flink

大数据

jdk

tcp/ip

本文分享自数据社微信公众号，前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体分享计划，欢迎热爱写作的你一起参与！

登录后参与评论

0 条评论

热度