Flink框架的安装和使用(二)

青旅半醒 2022-03-10 16:56 260阅读 0赞

**第一步.安装Flink**

1.我个人是在Mac电脑上进行安装的,安装过程很简单不需要做其他什么配置,下面就开始进行安装(有点慢):

brew install apache-flink

2.安装完成之后启动Flink和进入Web交互页面

brew info apache-flink

3.如果启动成功的话通过http://localhost:8081进入Web页面  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZ2hfc3Fo_size_16_color_FFFFFF_t_70]

**第二步.创建maven项目**

1.首先新建一个maven项目,然后点击新建一个骨架,每次使用的时候可以通过该骨架进行创建项目,如下图:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZ2hfc3Fo_size_16_color_FFFFFF_t_70 1]  
2.项目创建完成之后,可以看到在包路径下自动生成了四个java文件,打开WordCount文件可以看到如下代码:

public class WordCount {
    
    public static void main(String[] args) throws Exception {
    
    	// set up the execution environment
    	final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    
    	// get input data
    	DataSet<String> text = env.fromElements(
    			"To be, or not to be,--that is the question:--",
    			"Whether 'tis nobler in the mind to suffer",
    			"The slings and arrows of outrageous fortune",
    			"Or to take arms against a sea of troubles,"
    			);
    
    	DataSet<Tuple2<String, Integer>> counts =
    			// split up the lines in pairs (2-tuples) containing: (word,1)
    			text.flatMap(new LineSplitter())
    			// group by the tuple field "0" and sum up tuple field "1"
    			.groupBy(0)
    			.sum(1);
    
    	// execute and print result
    	counts.print();
    
    }
    
    //
    // 	User Functions
    //
    
    /**
     * Implements the string tokenizer that splits sentences into words as a user-defined
     * FlatMapFunction. The function takes a line (String) and splits it into
     * multiple pairs in the form of "(word,1)" (Tuple2<String, Integer>).
     */
    public static final class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
    
    	@Override
    	public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
    		// normalize and split the line
    		String[] tokens = value.toLowerCase().split("\\W+");
    
    		// emit the pairs
    		for (String token : tokens) {
    			if (token.length() > 0) {
    				out.collect(new Tuple2<String, Integer>(token, 1));
    			}
    		}
    	}
     }
    }

直接运行启动看是否能计算出结果;  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZ2hfc3Fo_size_16_color_FFFFFF_t_70 2]  
如果运行出上面的结果就可以对项目进行打包编译,然后通过后台进行启动,然后在Web界面总查看效果;

**第三步.打包编译和启动**

1.在根目录下执行打包编译命令

mvn clean package -Dmaven.test.skip=true

2.通过Flink框架运行jar包

进入到Flink安装的bin目录下执行以下命令

./flink run -c com.ligh.SocketTextStreamWordCount /Users/fish/Desktop/flinkdemo/target/flink-demo-1.0-SNAPSHOT.jar 127.0.0.1 9000

3.观察Web界面是否能监听到该端口的活动  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZ2hfc3Fo_size_16_color_FFFFFF_t_70 3]  
从上图中可以看出已经有一个正在运行中的进程了,可以点进去,显示结果如下:

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZ2hfc3Fo_size_16_color_FFFFFF_t_70 4]

到这里就完成了Flink框架的入门使用,总的来说该框架还是很简单上手的,以后我还会更深入的研究。

前进的道路上当你迷茫的时候,一定要坚信多努力一点未来就会好过很多 --------李国辉

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZ2hfc3Fo_size_16_color_FFFFFF_t_70]: /images/20220310/75086050c14249c5b63636918e3b7b92.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZ2hfc3Fo_size_16_color_FFFFFF_t_70 1]: /images/20220310/cc83291af1354678aafaf4f91be661cd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZ2hfc3Fo_size_16_color_FFFFFF_t_70 2]: /images/20220310/6627647ff43f4a7d97a94179986bfc25.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZ2hfc3Fo_size_16_color_FFFFFF_t_70 3]: /images/20220310/592362b1c77b4ec5aa7d8e160ae79e4a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZ2hfc3Fo_size_16_color_FFFFFF_t_70 4]: /images/20220310/cc3fc51b95a44e66b4581c48d0d19da6.png