01、flink的demo之旅

标签：macbook   session   包名   信息   ssi   规格   mit   软件   ide   

1. 安装filnk

由于是mac,所以直接使用brew工具下载软件，执行以下命令
brew install apache-flink

2. 启动flink,先进去bin目录,然后启动flink

cd /usr/local/Cellar/apache-flink/1.10.0/libexec/bin 
./start-cluster.sh 

有如下消息

Starting cluster.
[INFO] 1 instance(s) of standalonesession are already running on fangdadadeMacBook-Pro.local.
Starting standalonesession daemon on host fangdadadeMacBook-Pro.local.
[INFO] 1 instance(s) of taskexecutor are already running on fangdadadeMacBook-Pro.local.
Starting taskexecutor daemon on host fangdadadeMacBook-Pro.local.

打开浏览器：localhost:8081

3. maven构建本地项目，在工作目录下

mvn archetype:generate     -DarchetypeGroupId=org.apache.flink     -DarchetypeArtifactId=flink-quickstart-java     -DarchetypeVersion=1.7.2     -DgroupId=flink-project     -DartifactId=flink-project     -Dversion=0.1     -Dpackage=myflink     -DinteractiveMode=false

4. 编写本地任务

package com.myflink.demo;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * @author fangdada
 * @date 2020/7/7 10:19 下午
 * @Version 1.0
 */
public class FlinkTest {

    public static void main(String[] args) throws Exception {
        // 设置执行环境
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 获取输入数据
        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 =
                // 切割
                text.flatMap(new LineSplitter())
                // 分组
                .groupBy(0)
                // 求和
                .sum(1);

        counts.print();
    }

    public static final class LineSplitter implements FlatMapFunction<String,Tuple2<String,Integer>>{

        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
            // 规格化并拆分线条
            String[] tokens = value.toLowerCase().split("\\W+");

            // 取出对
            for (String token : tokens) {
                if(token.length() > 0){
                    out.collect(new Tuple2<>(token,1));
                }
            }

        }
    }
}

5. maven生命周期工具，直接编译打包

6. 提交任务到flink,首先还是进入bin目录，执行以下命令

-c 指定main方法的入口，注意，这个是类的全限定名，后面接上jar包名称

flink run -c com.myflink.demo.FlinkTest /Users/fangdada/IdeaProjects/flink-project/target/flink-project-0.1.jar         

运行信息如下，和我们在idea中运行的结果一致

fangdada@fangdadadeMacBook-Pro bin % flink run -c com.myflink.demo.FlinkTest /Users/fangdada/IdeaProjects/flink-project/target/flink-project-0.1.jar         
Job has been submitted with JobID f0c6005132f60e272a288ded111c04c5
Program execution finished
Job with JobID f0c6005132f60e272a288ded111c04c5 has finished.
Job Runtime: 307 ms
Accumulator Results: 
- 82dd9ba89434f13da6e012164f668d93 (java.util.ArrayList) [26 elements]


(a,1)
(against,1)
(and,1)
(arms,1)
(arrows,1)
(be,2)
(fortune,1)
(in,1)
(is,1)
(mind,1)
(nobler,1)
(not,1)
(of,2)
(or,2)
(outrageous,1)
(question,1)
(sea,1)
(slings,1)
(suffer,1)
(take,1)
(that,1)
(the,3)
(tis,1)
(to,4)
(troubles,1)
(whether,1)
fangdada@fangdadadeMacBook-Pro bin % 

7. 在查看flink客户端

01、flink的demo之旅

标签：macbook   session   包名   信息   ssi   规格   mit   软件   ide   

原文地址：https://www.cnblogs.com/fangdadada/p/13289071.html