Flink通过flink-yarn远程提交任务到yarn集群

港控/mmm° 2022-09-14 01:47 551阅读 0赞

public void crateStreamTaskByFlinkClient() {
        //flink的本地配置目录，为了得到flink的配置
        // 如果出现org.apache.flink.streaming.runtime.tasks.StreamTaskException: Cannot instantiate user function.错误
        // 则在flink-config.yaml加入
        // classloader.resolve-order: parent-first
        String configurationDirectory = "/opt/module/flink-1.11.4/conf";
        // String configurationDirectory = "/home/lxj/workspace/Olt-Test/bigdata/bigdataserver/src/main/resources/flink/conf";
    
        //存放flink集群相关的jar包目录
        String flinkLibs = "hdfs://hadoop113:8020/jar/flink11/libs";
        //用户jar
        String userJarPath = "hdfs://hadoop113:8020/jar/userTask/Flink1.11-1.0-SNAPSHOT-jar-with-dependencies.jar";
        String flinkDistJar = "hdfs://hadoop113:8020/jar/flink11/libs/flink-dist_2.12-1.11.4.jar";
    
        YarnClient yarnClient = YarnClient.createYarnClient();
        YarnConfiguration yarnConfiguration = new YarnConfiguration();
        yarnClient.init(yarnConfiguration);
        yarnClient.start();
    
        // 设置日志的，没有的话看不到日志
        YarnClusterInformationRetriever clusterInformationRetriever = YarnClientYarnClusterInformationRetriever
            .create(yarnClient);
    
        //获取flink的配置
        Configuration flinkConfiguration = GlobalConfiguration.loadConfiguration(
            configurationDirectory);
    
        flinkConfiguration.set(CheckpointingOptions.INCREMENTAL_CHECKPOINTS, true);
    
        flinkConfiguration.set(
            PipelineOptions.JARS,
            Collections.singletonList(userJarPath));
    
        Path remoteLib = new Path(flinkLibs);
        flinkConfiguration.set(
            YarnConfigOptions.PROVIDED_LIB_DIRS,
            Collections.singletonList(remoteLib.toString()));
    
        flinkConfiguration.set(
            YarnConfigOptions.FLINK_DIST_JAR,
            flinkDistJar);
    
        // 设置为application模式
        flinkConfiguration.set(
            DeploymentOptions.TARGET,
            YarnDeploymentTarget.APPLICATION.getName());
    
        // yarn application name
        flinkConfiguration.set(YarnConfigOptions.APPLICATION_NAME, "flink-application");
    
        YarnLogConfigUtil.setLogConfigFileInConfig(flinkConfiguration, configurationDirectory);
    
        ClusterSpecification clusterSpecification = new ClusterSpecification.ClusterSpecificationBuilder()
            .createClusterSpecification();
    
        // 设置用户jar的参数和主类
        ApplicationConfiguration appConfig = new ApplicationConfiguration(new String[] {"test"}, "com.starnet.server.bigdata.flink.WordCount");
    
        YarnClusterDescriptor yarnClusterDescriptor = new YarnClusterDescriptor(
            flinkConfiguration,
            yarnConfiguration,
            yarnClient,
            clusterInformationRetriever,
            true);
    
        try {
            ClusterClientProvider<ApplicationId> clusterClientProvider = yarnClusterDescriptor.deployApplicationCluster(
                clusterSpecification,
                appConfig);
    
            ClusterClient<ApplicationId> clusterClient = clusterClientProvider.getClusterClient();
    
            ApplicationId applicationId = clusterClient.getClusterId();
            String webInterfaceURL = clusterClient.getWebInterfaceURL();
            log.error("applicationId is {}", applicationId);
            log.error("webInterfaceURL is {}", webInterfaceURL);
            
            // 退出
            // yarnClusterDescriptor.killCluster(applicationId);
        } catch (Exception e){
            log.error(e.getMessage(), e);
        }
    }