flink on yarn 远程提交

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.client.cli.CliFrontend;
import org.apache.flink.client.cli.CustomCommandLine;
import org.apache.flink.client.cli.DefaultCLI;
import org.apache.flink.client.cli.GenericCLI;
import org.apache.flink.client.deployment.ClusterDeploymentException;
import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.client.deployment.application.ApplicationConfiguration;
import org.apache.flink.client.program.ClusterClientProvider;
import org.apache.flink.configuration.*;
import org.apache.flink.runtime.security.SecurityConfiguration;
import org.apache.flink.runtime.security.SecurityUtils;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.yarn.YarnClientYarnClusterInformationRetriever;
import org.apache.flink.yarn.YarnClusterDescriptor;
import org.apache.flink.yarn.YarnClusterInformationRetriever;
import org.apache.flink.yarn.configuration.YarnConfigOptions;
import org.apache.flink.yarn.configuration.YarnDeploymentTarget;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.junit.Test;

import java.io.File;
import java.lang.reflect.Constructor;
import java.lang.reflect.UndeclaredThrowableException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Optional;

import static org.apache.flink.util.Preconditions.checkNotNull;


@Slf4j
public class AppTestV1 {


    @Test
    public void submitJobWithYarnDesc() throws ClusterDeploymentException {
        // hadoop
        String hadoopConfDir = "C:\\Users\\HPN-21-117\\software\\configClusterhadoop\\configurations\\cdh5";
//        String hadoopConfDir = "C:\\Users\\HPN-21-117\\software\\configClusterhadoop\\configurations\\cdh6";
        //flink的本地配置目录，为了得到flink的配置
        String flinkConfDir = "C:\\Users\\HPN-21-117\\software\\flink-1.12.7\\conf";
        //存放flink集群相关的jar包目录
        String flinkLibs = "hdfs://cdh-node0.hypers.com:8020/flink/1.12.7/lib";
        //用户jar
        String userJarPath =  "hdfs://cdh-node0.hypers.com:8020/flink/demo/TopSpeedWindowing.jar";
        String flinkDistJar = "hdfs://cdh-node0.hypers.com:8020/flink/1.12.7/lib/flink-dist_2.12-1.12.7.jar";
//        String flinkDistJar = "hdfs://cdh-node0.hypers.com:8022/flink/lib";
        String[] args = "".split("\\s+");
        String appMainClass = "org.apache.flink.streaming.examples.windowing.TopSpeedWindowing";

        YarnClient yarnClient = YarnUtils.getYarnClient(hadoopConfDir);
        yarnClient.start();

        Configuration flinkConf = GlobalConfiguration.loadConfiguration(flinkConfDir);
        //set run model
        flinkConf.setString(DeploymentOptions.TARGET, YarnDeploymentTarget.APPLICATION.getName());
        //set application name
        flinkConf.setString(YarnConfigOptions.APPLICATION_NAME, "onYarnApiSubmitCase");
        //flink on yarn dependency
        flinkConf.set(YarnConfigOptions.PROVIDED_LIB_DIRS, Collections.singletonList(new Path(flinkLibs).toString()));
        flinkConf.set(YarnConfigOptions.FLINK_DIST_JAR, flinkDistJar);
        flinkConf.set(PipelineOptions.JARS, Collections.singletonList(new Path(userJarPath).toString()));
        //设置：资源/并发度
        flinkConf.setInteger(CoreOptions.DEFAULT_PARALLELISM, 1);
        flinkConf.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.parse("1G"));
        flinkConf.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.parse("1G"));
        flinkConf.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, 1);


        ClusterSpecification clusterSpecification = new ClusterSpecification
                .ClusterSpecificationBuilder()
                .setMasterMemoryMB(1024)
                .setTaskManagerMemoryMB(1024)
                .setSlotsPerTaskManager(2)
                .createClusterSpecification();

        YarnClusterInformationRetriever ycir = YarnClientYarnClusterInformationRetriever.create(yarnClient);

        YarnConfiguration yarnConf = (YarnConfiguration) yarnClient.getConfig();

        ApplicationConfiguration appConfig = new ApplicationConfiguration(args, appMainClass);

        YarnClusterDescriptor yarnClusterDescriptor = new YarnClusterDescriptor(
                flinkConf,
                yarnConf,
                yarnClient,
                ycir,
                false);

        ClusterClientProvider<ApplicationId> applicationCluster =
                yarnClusterDescriptor.deployApplicationCluster( clusterSpecification, appConfig );

        yarnClient.stop();

    }

    @Test
    public  void submitJobWithCliForte() throws Exception {

        System.setProperty("ENV_FLINK_CONF_DIR", "C:\\Users\\HPN-21-117\\software\\flink-1.14.6\\conf");
        System.setProperty("FLINK_CONF_DIR", "C:\\Users\\HPN-21-117\\software\\flink-1.14.6\\conf");

        // 1. find the configuration directory
        final String configurationDirectory = getConfigurationDirectoryFromDir("C:\\Users\\HPN-21-117\\software\\flink-1.14.6\\conf");

        // 2. load the global configuration
        final Configuration configuration =
                GlobalConfiguration.loadConfiguration(configurationDirectory);

        // 3. load the custom command lines
        final List<CustomCommandLine> customCommandLines =
                loadCustomCommandLines(configuration, configurationDirectory);

        try {
            final CliFrontend cli = new CliFrontend(configuration, customCommandLines);

            SecurityUtils.install(new SecurityConfiguration(cli.getConfiguration()));
            String[] args = "run-application -t yarn-application hdfs://cdh-node0.hypers.com:8022/flink/demo/TopSpeedWindowing.jar".split("\\s+");
            int retCode =
                    SecurityUtils.getInstalledContext().runSecured(() -> cli.parseAndRun(args));
            System.exit(retCode);
        } catch (Throwable t) {
            final Throwable strippedThrowable =
                    ExceptionUtils.stripException(t, UndeclaredThrowableException.class);
            log.error("Fatal error while running command line interface.", strippedThrowable);
            strippedThrowable.printStackTrace();
            System.exit(31);
        }

    }

    public static List<CustomCommandLine> loadCustomCommandLines(
            Configuration configuration, String configurationDirectory) {
        List<CustomCommandLine> customCommandLines = new ArrayList<>();
        customCommandLines.add(new GenericCLI(configuration, configurationDirectory));

        //	Command line interface of the YARN session, with a special initialization here
        //	to prefix all options with y/yarn.
        final String flinkYarnSessionCLI = "org.apache.flink.yarn.cli.FlinkYarnSessionCli";
        try {
            customCommandLines.add(
                    loadCustomCommandLine(
                            flinkYarnSessionCLI,
                            configuration,
                            configurationDirectory,
                            "y",
                            "yarn"));
        } catch (NoClassDefFoundError | Exception e) {
            final String errorYarnSessionCLI = "org.apache.flink.yarn.cli.FallbackYarnSessionCli";
            try {
                log.info("Loading FallbackYarnSessionCli");
                customCommandLines.add(loadCustomCommandLine(errorYarnSessionCLI, configuration));
            } catch (Exception exception) {
                log.warn("Could not load CLI class {}.", flinkYarnSessionCLI, e);
            }
        }

        //	Tips: DefaultCLI must be added at last, because getActiveCustomCommandLine(..) will get
        // the
        //	      active CustomCommandLine in order and DefaultCLI isActive always return true.
        customCommandLines.add(new DefaultCLI());

        return customCommandLines;
    }

    /**
     * Loads a class from the classpath that implements the CustomCommandLine interface.
     *
     * @param className The fully-qualified class name to load.
     * @param params The constructor parameters
     */
    private static CustomCommandLine loadCustomCommandLine(String className, Object... params)
            throws Exception {

        Class<? extends CustomCommandLine> customCliClass =
                Class.forName(className).asSubclass(CustomCommandLine.class);

        // construct class types from the parameters
        Class<?>[] types = new Class<?>[params.length];
        for (int i = 0; i < params.length; i++) {
            checkNotNull(params[i], "Parameters for custom command-lines may not be null.");
            types[i] = params[i].getClass();
        }

        Constructor<? extends CustomCommandLine> constructor = customCliClass.getConstructor(types);

        return constructor.newInstance(params);
    }

    public static String getConfigurationDirectoryFromDir(String env_flink_conf_dir) {

//        String env_flink_conf_dir = System.getenv(flinkConfDir);

        String location = Optional.ofNullable(env_flink_conf_dir)
                .filter(dir -> new File(dir).exists())
                .orElseThrow(() -> new RuntimeException(
                        "The configuration directory '"
                                + env_flink_conf_dir
                                + "', specified in the '"
                                + ConfigConstants.ENV_FLINK_CONF_DIR
                                + "' environment variable, does not exist."));

        return location;
    }

    public static String getConfigurationDirectoryFromEnv() {
        System.setProperty("MY_VAR", "value");

        String env_flink_conf_dir = System.getenv(ConfigConstants.ENV_FLINK_CONF_DIR);

        String location = Optional.ofNullable(env_flink_conf_dir)
                .filter(dir -> new File(dir).exists())
                .orElseThrow(() -> new RuntimeException(
                        "The configuration directory '"
                                + env_flink_conf_dir
                                + "', specified in the '"
                                + ConfigConstants.ENV_FLINK_CONF_DIR
                                + "' environment variable, does not exist."));

        return location;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229