StreamX 部署 Flink Stream 应用

• 创建工程
• 在 pom 文件中添加 Flink 相关依赖
• 示例代码
• 代码推送到 gitee
• 配置项目
• 编译项目
• 提交应用
• Yarn 平台确认执行结果

笔者工程名叫Flink-StreamX

    
        8
        8
        1.13.6
        2.11
        1.7.30
    
    
        
            org.apache.flink
            flink-java
            ${flink.version}
            provided
        
        
            org.apache.flink
            flink-streaming-java_${scala.binary.version}
            ${flink.version}
            provided
        
        
            org.apache.flink
            flink-clients_${scala.binary.version}
            ${flink.version}
            provided
        
        
            org.apache.flink
            flink-runtime-web_${scala.binary.version}
            ${flink.version}
            provided
        
        
            org.slf4j
            slf4j-api
            ${slf4j.version}
            provided
        
        
            org.slf4j
            slf4j-log4j12
            ${slf4j.version}
            provided
        
        
            org.apache.logging.log4j
            log4j-to-slf4j
            2.14.0
            provided
        
    
    
        
            
                org.apache.maven.plugins
                maven-shade-plugin
                3.2.4
                
                    
                        package
                        
                            shade
                        
                        
                            
                                
                                    com.google.code.findbugs:jsr305
                                    org.slf4j:*
                                    log4j:*
                                
                            
                            
                                
                                    
                                    *:*
                                    
                                        META-INF/*.SF
                                        META-INF/*.DSA
                                        META-INF/*.RSA
                                    
                                
                            
                            
                                
                                
                            
                        
                    
                
            
        
    

package com.apache.bigdata;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class UnboundedWC {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env.socketTextStream("hadoop102", 9999)
                .flatMap(new FlatMapFunction() {
                    @Override
                    public void flatMap(String line, Collector out) throws Exception {
                        for (String word : line.split(" ")) {
                            out.collect(word);
                        }
                    }
                })
                .map(new MapFunction>() {
                    @Override
                    public Tuple2 map(String word) throws Exception {
                        return Tuple2.of(word, 1l);
                    }
                })
                .keyBy(new KeySelector, String>() {
                    @Override
                    public String getKey(Tuple2 t) throws Exception {
                        return t.f0; // t._1
                    }
                })
                .sum(1)
                .print();
                
        env.execute();
    }
}

在 streamx 平台部署应用的时候要求代码最好部署在 git 平台，比如 github 或 gitee。作为国内用户我们选择比较稳定的 gitee。

如果不会git/gitee的小伙伴，可以从基础学习一下，时间不长，一天足够。

我的项目推送地址：https://gitee.com/luan_hao/Flink-StreamX/

编译前：

第一次编译需要的时间比较久, 因为需要下载许多的依赖。

编译成功后：

1）创建应用

2）配置应用

3）上线应用

4）启动应用（注意先启动 socket: nc -lk 9999）

启动成功的表现：

开始运行：

在弹出的界面里点击Task Managers，然后点击正在运行的任务

[root@hadoop102 local]# nc -lk 9999
spark kafka
flink hadoop

再点击stdout即可查看输出：

成功。