亲宝软件园·资讯

展开

轻装上阵Flink--在IDEA上开发基于Flink的实时数据流程序

壮壮熊 人气:2

前言

      本文介绍如何在IDEA上快速开发基于Flink框架的DataStream程序。先直接上手!

环境清单

      案例是在win7运行。安装VirtualBox,在VirtualBox上安装Centos操作系统。所有资源都在百度云上,有需要请直接下载。安装教程基本都是傻瓜式,文章不做讲述,有需要直接网上搜索。

资源 版本
VirtualBox 5.2.16
Centos 6.5
Maven 3.6.3
JDK 8u241
IDEA 2019.3.2
Flink 1.10.0

链接:http://pan.baidu.com/s/12rXlY_z_Fck8-NRXdZ5row

提取码:qt2p

轻装上阵

1、IP设置

      Centos的设置静态IP为192.168.2.20,关闭防火墙

1 vi /etc/sysconfig/network-scripts/ifcfg-eth0
2 DEVICE=eth0
3 TYPE=Ethernet
4 ONBOOT=yes #开机启动eth0网卡
5 NM_CONTROLLED=yes
6 BOOTPROTO=static
7 IPADDR=192.168.2.20
8 GATEWAY=192.168.2.1
9 NETMASK=255.255.255.0
     如果此时ping www.baidu.com等不通,需要我们添加dns服务器。
1 [root@localhost network-scripts]# vi /etc/resolv.conf
2 nameserver 192.168.2.1
  重新启动网络服务
1 [root@localhost network-scripts]# service network restart                   
2 正在关闭接口 eth0:[确定]
3 关闭环回接口:[确定]
4 弹出环回接口:[确定]
5 弹出界面 eth0:Determining if ip address 192.168.2.20 is already in use for device eth0...
6                                                            [确定]
      关闭防火墙
1 [root@localhost network-scripts]# service iptables stop

2、创建项目

   在win7的命令行下,用mvn命令创建开发模板

1 mvn archetype:generate -DarchetypeGroupId=org.apache.flink -DarchetypeArtifactId=flink-quickstart-java -DarchetypeVersion=1.10.0
这种方式允许你为新项目命名。它将以交互式的方式询问你项目的 groupId、artifactId 和 package 名称。
用tree命令看下,如下结构。项目是一个 Maven project,它包含了两个类:StreamingJob 和 BatchJob
分别是 DataStream and DataSet 程序的基础骨架程序。main 方法是程序的入口,既可用于IDE测试/执行,也可用于部署。
 1 │  pom.xml
 2 └─src
 3     └─main
 4         ├─java
 5         │  └─com
 6         │      └─ryan
 7         │              BatchJob.java
 8         │              StreamingJob.java
 9         └─resources
10                 log4j.properties
3、写一个自己的DataStream的程序

 功能介绍:WindowWordCount.java,5s为一个时间窗口,摄取数据源的数据,计算单词出现的次数。

 实时数据流计算简易架构图:

为了演示方便,这里我们只演示消息队列和Flink Job两个模块,利用nc工具来替代消息队列作为Flink Job摄取的数据源。

代码:

 1 package com.ryan;
 2 import org.apache.flink.api.common.functions.FlatMapFunction;
 3 import org.apache.flink.api.java.tuple.Tuple2;
 4 import org.apache.flink.streaming.api.datastream.DataStream;
 5 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 6 import org.apache.flink.streaming.api.windowing.time.Time;
 7 import org.apache.flink.util.Collector;
 8 public class WindowWordCount {
 9     public static void main(String[] args) throws Exception {
10         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
11         DataStream<Tuple2<String, Integer>> dataStream = env
12                 .socketTextStream("192.168.2.20", 9999)
13                 .flatMap(new Splitter())
14                 .keyBy(0)
15                 .timeWindow(Time.seconds(5))
16                 .sum(1);
17         dataStream.print();
18         env.execute("Window WordCount");
19     }
20     public static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
21         @Override
22         public void flatMap(String sentence, Collector<Tuple2<String, Integer>> out) throws Exception {
23             for (String word: sentence.split(" ")) {
24                 out.collect(new Tuple2<String, Integer>(word, 1));
25             }
26         }
27     }
28 }

在centos机器上,命令行启动nc

1 nc -lk 9999

IDEA上直接run main方法,然后在centos机器上,不断输入单词。

1 [ryan@localhost ~]$ nc -lk 9999
2 java
3 java
4 shen
5 深圳 深圳
IDEA控制台上输出如下:

注意:第一次在IDEA上运行这个程序,可能会报如下异常

1 java.lang.NoClassDefFoundError: org/apache/flink/streaming/apihttps://img.qb5200.com/download-x/datastream/DataStream

原因是IDEA没有导入flink 的lib下的jar包。导入即可。

4、打包发布到centos平台上的Flink集群

      修改pom.xml文件的mainclass的值为com.ryan.WindowWordCount

1 <mainClass>com.ryan.WindowWordCount</mainClass>

      执行mvn clean install,得到flink-demo-1.0-SNAPSHOT.jar,并上传到centos机器上。

1 mvn clean install

      打开两个centos的控制台,一个用于打开nc,一个用于运行我们打包好的Flink jar包。

1 [ryan@localhost ~]$ nc -lk 9999
2 java
3 shen
4 深圳 深圳 深圳
1 [root@localhost flink-1.10.0]# bin/flink run flink-demo/flink-demo-1.0-SNAPSHOT.jar 
2 Job has been submitted with JobID 9931a9dfc2eddeb2d0b5ed15578bd488
  回到win7上,用浏览器打开http://192.168.2.20:8081/,在Running Jobs上,可以看到一条记录。

 

       在Task Managers上,Stdout模块看到程序输出的结果。

       所有代码都上传到github上,有需要的朋友可以下载

1 https://github.com/qinxiongzhou/flink-demo

       至此,我们完成了开发编译调试到最终上线生产运行。喜欢请关注公众号--程序猿牧场,谢谢!

 

加载全部内容

相关教程
猜你喜欢
用户评论