flink on yarn with kerberos 边缘提交_flink提交yarn配置带kerberos

flink on yarn 带kerberos 远程提交 实现

1. flink kerberos 配置
   
2. 先使用ugi进行一次认证
3. 正常提交

import com.google.common.io.Files;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FileUtils;
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.security.UserGroupInformation;
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.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.UndeclaredThrowableException;
import java.net.MalformedURLException;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;

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


/**
* @author: jiayeli.cn
* @description
* @date: 2023/8/29 下午9:09
*/

@Slf4j
public class YarnClientTestCase {

   @Test
   public void submitJobWithYarnDesc() throws ClusterDeploymentException, IOException {
       // hadoop
       String hadoopConfDir = "/x/x/software/spark-3.3.2-bin-hadoop3/etc/hadoop";
       //flink的本地配置目录，为了得到flink的配置
       String flinkConfDir = "/opt/flink-1.14.3/conf";
       //存放flink集群相关的jar包目录
       String flinkLibs = "hdfs://node01:8020/lib/flink";
       //用户jar
       String userJarPath =  "hdfs://node01:8020/jobs/streaming/testCase/TopSpeedWindowing.jar";
       String flinkDistJar = "hdfs://node01:8020/lib/flink/flink-dist_2.12-1.14.3.jar";
       String[] args = "".split("\\s+");
       String appMainClass = "org.apache.flink.streaming.examples.windowing.TopSpeedWindowing";
       String principal = "dev@JIAYELI.COM";
       String keyTab = "/x/x/workspace/bigdata/sparkLauncherTestcase/src/test/resource/dev_uer.keytab";

       enableKrb5(principal, keyTab);
       YarnClient yarnClient = YarnClient.createYarnClient();
       YarnConfiguration yarnConfiguration = new YarnConfiguration();
       Optional.ofNullable(hadoopConfDir)
           .map(e -> new File(e))
           .filter(dir -> dir.exists())
           .map(File::listFiles)
           .ifPresent(files -> {
               Arrays.asList(files).stream()
                       .filter(file -> Files.getFileExtension(file.getName()).equals(".xml"))
                       .forEach(conf -> yarnConfiguration.addResource(conf.getPath()));
           });

       yarnClient.init(yarnConfiguration);
       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();

   }

   private void enableKrb5(String principal, String keyTab) throws IOException {
     System.setProperty("java.security.krb5.conf", "/x/x/Documents/kerberos/krb5.conf");

       org.apache.hadoop.conf.Configuration krb5conf = new org.apache.hadoop.conf.Configuration();


       String krb5ConfPath = "/x/x/Documents/kerberos/krb5.conf";

       krb5conf.set("hadoop.security.authentication", "kerberos");

       //      UserGroupInformation.setConfiguration(conf)
       UserGroupInformation.setConfiguration(krb5conf);

       // 登录Kerberos并获取UserGroupInformation实例
       UserGroupInformation.loginUserFromKeytab(principal, keyTab);
       UserGroupInformation ugi = UserGroupInformation.getCurrentUser();

       log.debug(ugi.toString());
   }
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