分布式计算引擎 Flink/Spark on k8s 的实现对比以及实践

$ ./bin/spark-submit \
    --master k8s://https://<k8s-apiserver-host>:<k8s-apiserver-port> \
    --deploy-mode cluster \
    --name spark-pi \
    --class org.apache.spark.examples.SparkPi \
    --conf spark.executor.instances=5 \
    --conf spark.kubernetes.container.image=<spark-image> \
    local:///path/to/examples.jar

...
--packages org.apache.hadoop:hadoop-aws:3.2.0
--conf spark.kubernetes.file.upload.path=s3a://<s3-bucket>/path
--conf spark.hadoop.fs.s3a.access.key=...
--conf spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem
--conf spark.hadoop.fs.s3a.fast.upload=true
--conf spark.hadoop.fs.s3a.secret.key=....
--conf spark.driver.extraJavaOptions=-Divy.cache.dir=/tmp -Divy.home=/tmp
file:///full/path/to/app.jar

apiVersion: batch/v1
kind: Job
metadata:
  name: hello
spec:
  template:
    # 下面的是一个 pod template
    spec:
      containers:
      - name: hello
        image: busybox
        command: ['sh', '-c', 'echo "Hello, Kubernetes!" && sleep 3600']
      restartPolicy: OnFailure
    # The pod template ends here

$ kubectl create serviceaccount spark
$ kubectl create clusterrolebinding spark-role --clusterrole=edit --serviceaccount=spark:spark --namespace=spark

$ spark-submit --master k8s://https://<api-server>:<port> --deploy-mode cluster --class org.apache.spark.examples.SparkPiSleep --conf spark.executor.memory=2g --conf spark.driver.memory=2g --conf spark.driver.core=1 --conf spark.app.name=test12 --conf spark.kubernetes.submission.waitAppCompletion=false --conf spark.executor.core=1 --conf spark.kubernetes.container.image=<image-id> --conf spark.eventLog.enabled=false --conf spark.shuffle.service.enabled=false --conf spark.executor.instances=1 --conf spark.dynamicAllocation.enabled=false --conf spark.kubernetes.namespace=spark --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark --conf spark.executor.core=1  local:///path/to/main/jar

$ kubectl get po -n spark
NAME                               READY   STATUS              RESTARTS   AGE
spark-pi-5b88a27b576050dd-exec-1   0/1     ContainerCreating   0          2s
test12-9fd3c27b576039ae-driver     1/1     Running             0          8s

$ kubectl get svc -n spark
NAME                                 TYPE           CLUSTER-IP     EXTERNAL-IP     PORT(S)                                       AGE
test12-9fd3c27b576039ae-driver-svc   ClusterIP      None           <none>          7078/TCP,7079/TCP,4040/TCP                    110s

$ kubectl get svc test12-9fd3c27b576039ae-driver-svc -n spark -oyaml
apiVersion: v1
kind: Service
metadata:
  creationTimestamp: "2021-08-18T03:48:50Z"
  name: test12-9fd3c27b576039ae-driver-svc
  namespace: spark
  # service 的 ownerReference 指向了 driver pod，只要 driver pod 被删除，该 service 也会被删除
  ownerReferences:
  - apiVersion: v1
    controller: true
    kind: Pod
    name: test12-9fd3c27b576039ae-driver
    uid: 56a50a66-68b5-42a0-b2f6-9a9443665d95
  resourceVersion: "9975441"
  uid: 06c1349f-be52-4133-80d9-07af34419b1f

$ ./bin/flink run-application \
    --target kubernetes-application \
    -Dkubernetes.cluster-id=my-first-application-cluster \
    -Dkubernetes.container.image=custom-image-name \
    local:///opt/flink/usrlib/my-flink-job.jar

// Set the cluster manager
    val clusterManager: Int = args.master match {
      case "yarn" => YARN
      case m if m.startsWith("spark") => STANDALONE
      case m if m.startsWith("mesos") => MESOS
      case m if m.startsWith("k8s") => KUBERNETES
      case m if m.startsWith("local") => LOCAL
      case _ =>
        error("Master must either be yarn or start with spark, mesos, k8s, or local")
        -1
    }

private[spark] class KubernetesClientApplication extends SparkApplication {
override def start(args: Array[String], conf: SparkConf): Unit = {

val parsedArguments = ClientArguments.fromCommandLineArgs(args)

run(parsedArguments, conf)

}
private def run(clientArguments: ClientArguments, sparkConf: SparkConf): Unit = {

// For constructing the app ID, we can't use the Spark application name, as the app ID is going

// to be added as a label to group resources belonging to the same application. Label values are

// considerably restrictive, e.g. must be no longer than 63 characters in length. So we generate

// a unique app ID (captured by spark.app.id) in the format below.

val kubernetesAppId = KubernetesConf.getKubernetesAppId()

val kubernetesConf = KubernetesConf.createDriverConf(

sparkConf,

kubernetesAppId,

clientArguments.mainAppResource,

clientArguments.mainClass,

clientArguments.driverArgs,

clientArguments.proxyUser)

// The master URL has been checked for validity already in SparkSubmit.

// We just need to get rid of the "k8s://" prefix here.

val master = KubernetesUtils.parseMasterUrl(sparkConf.get("spark.master"))

val watcher = new LoggingPodStatusWatcherImpl(kubernetesConf)
Utils.tryWithResource(SparkKubernetesClientFactory.createKubernetesClient(
  master,
  Some(kubernetesConf.namespace),
  KUBERNETES_AUTH_SUBMISSION_CONF_PREFIX,
  SparkKubernetesClientFactory.ClientType.Submission,
  sparkConf,
  None,
  None)) { kubernetesClient =>
    val client = new Client(
      kubernetesConf,
      new KubernetesDriverBuilder(),
      kubernetesClient,
      watcher)
    client.run()
}

}

}

private[spark] class Client(

conf: KubernetesDriverConf,

builder: KubernetesDriverBuilder,

kubernetesClient: KubernetesClient,

watcher: LoggingPodStatusWatcher) extends Logging {

def run(): Unit = {

// 构造 Driver 的 Pod

val resolvedDriverSpec = builder.buildFromFeatures(conf, kubernetesClient)

val configMapName = KubernetesClientUtils.configMapNameDriver

val confFilesMap = KubernetesClientUtils.buildSparkConfDirFilesMap(configMapName,

conf.sparkConf, resolvedDriverSpec.systemProperties)

val configMap = KubernetesClientUtils.buildConfigMap(configMapName, confFilesMap)
// 修改 Pod 的 container spec：增加 SPARK_CONF_DIR
val resolvedDriverContainer = new ContainerBuilder(resolvedDriverSpec.pod.container)
  .addNewEnv()
    .withName(ENV_SPARK_CONF_DIR)
    .withValue(SPARK_CONF_DIR_INTERNAL)
    .endEnv()
  .addNewVolumeMount()
    .withName(SPARK_CONF_VOLUME_DRIVER)
    .withMountPath(SPARK_CONF_DIR_INTERNAL)
    .endVolumeMount()
  .build()
val resolvedDriverPod = new PodBuilder(resolvedDriverSpec.pod.pod)
  .editSpec()
    .addToContainers(resolvedDriverContainer)
    .addNewVolume()
      .withName(SPARK_CONF_VOLUME_DRIVER)
      .withNewConfigMap()
        .withItems(KubernetesClientUtils.buildKeyToPathObjects(confFilesMap).asJava)
        .withName(configMapName)
        .endConfigMap()
      .endVolume()
    .endSpec()
  .build()
val driverPodName = resolvedDriverPod.getMetadata.getName

var watch: Watch = null
var createdDriverPod: Pod = null
try {
  // 通过 k8s client 创建 Driver Pod
  createdDriverPod = kubernetesClient.pods().create(resolvedDriverPod)
} catch {
  case NonFatal(e) =>
    logError("Please check \"kubectl auth can-i create pod\" first. It should be yes.")
    throw e
}
try {
  // 创建其他资源，修改 owner reference 等
  val otherKubernetesResources = resolvedDriverSpec.driverKubernetesResources ++ Seq(configMap)
  addOwnerReference(createdDriverPod, otherKubernetesResources)
  kubernetesClient.resourceList(otherKubernetesResources: _*).createOrReplace()
} catch {
  case NonFatal(e) =>
    kubernetesClient.pods().delete(createdDriverPod)
    throw e
}
val sId = Seq(conf.namespace, driverPodName).mkString(":")
// watch pod
breakable {
  while (true) {
    val podWithName = kubernetesClient
      .pods()
      .withName(driverPodName)
    // Reset resource to old before we start the watch, this is important for race conditions
    watcher.reset()
    watch = podWithName.watch(watcher)

    // Send the latest pod state we know to the watcher to make sure we didn't miss anything
    watcher.eventReceived(Action.MODIFIED, podWithName.get())

    // Break the while loop if the pod is completed or we don't want to wait
    // 根据参数 "spark.kubernetes.submission.waitAppCompletion" 判断是否需要退出
    if(watcher.watchOrStop(sId)) {
      watch.close()
      break
    }
  }
}

}

}

private def createTaskScheduler(...) = {

case masterUrl =>

// 创建出 KubernetesClusterManager

val cm = getClusterManager(masterUrl) match {

case Some(clusterMgr) => clusterMgr

case None => throw new SparkException("Could not parse Master URL: '" + master + "'")

}

try {

val scheduler = cm.createTaskScheduler(sc, masterUrl)

// 上面创建出来的 KubernetesClusterManager 这里会创建出 KubernetesClusterSchedulerBackend

val backend = cm.createSchedulerBackend(sc, masterUrl, scheduler)

cm.initialize(scheduler, backend)

(backend, scheduler)

} catch {

case se: SparkException => throw se

case NonFatal(e) =>

throw new SparkException("External scheduler cannot be instantiated", e)

}

}

// 方法 getClsuterManager 会通过 ServiceLoader 加载所有实现 ExternalClusterManager 的 ClusterManager (KubernetesClusterManager 和 YarnClusterManager)，然后通过 master url 进行 filter，选出 KubernetesClusterManager

private def getClusterManager(url: String): Option[ExternalClusterManager] = {

val loader = Utils.getContextOrSparkClassLoader

val serviceLoaders =

ServiceLoader.load(classOf[ExternalClusterManager], loader).asScala.filter(.canCreate(url))

if (serviceLoaders.size > 1) {

throw new SparkException(

s"Multiple external cluster managers registered for the url url: serviceLoaders")

}

serviceLoaders.headOption

}

private def requestNewExecutors(

expected: Int,

running: Int,

applicationId: String,

resourceProfileId: Int,

pvcsInUse: Seq[String]): Unit = {

val numExecutorsToAllocate = math.min(expected - running, podAllocationSize)

logInfo(s"Going to request $numExecutorsToAllocate executors from Kubernetes for " +

s"ResourceProfile Id: resourceProfileId, target: expected running: $running.")

// Check reusable PVCs for this executor allocation batch

val reusablePVCs = getReusablePVCs(applicationId, pvcsInUse)

for ( _ <- 0 until numExecutorsToAllocate) {

val newExecutorId = EXECUTOR_ID_COUNTER.incrementAndGet()

val executorConf = KubernetesConf.createExecutorConf(

conf,

newExecutorId.toString,

applicationId,

driverPod,

resourceProfileId)

// 构造 Executor 的 Pod Spec

val resolvedExecutorSpec = executorBuilder.buildFromFeatures(executorConf, secMgr,

kubernetesClient, rpIdToResourceProfile(resourceProfileId))

val executorPod = resolvedExecutorSpec.pod

val podWithAttachedContainer = new PodBuilder(executorPod.pod)

.editOrNewSpec()

.addToContainers(executorPod.container)

.endSpec()

.build()

val resources = replacePVCsIfNeeded(

podWithAttachedContainer, resolvedExecutorSpec.executorKubernetesResources, reusablePVCs)

// 创建 Executor Pod

val createdExecutorPod = kubernetesClient.pods().create(podWithAttachedContainer)

try {

// 增加 owner reference

addOwnerReference(createdExecutorPod, resources)

resources

.filter(.getKind == "PersistentVolumeClaim")

.foreach { resource =>

if (conf.get(KUBERNETES_DRIVER_OWN_PVC) && driverPod.nonEmpty) {

addOwnerReference(driverPod.get, Seq(resource))

}

val pvc = resource.asInstanceOf[PersistentVolumeClaim]

logInfo(s"Trying to create PersistentVolumeClaim ${pvc.getMetadata.getName} with " +

s"StorageClass ${pvc.getSpec.getStorageClassName}")

kubernetesClient.persistentVolumeClaims().create(pvc)

}

newlyCreatedExecutors(newExecutorId) = (resourceProfileId, clock.getTimeMillis())

logDebug(s"Requested executor with id $newExecutorId from Kubernetes.")

} catch {

case NonFatal(e) =>

kubernetesClient.pods().delete(createdExecutorPod)

throw e

}

}

}

public interface ClusterDescriptor<T> extends AutoCloseable {

/* Returns a String containing details about the cluster (NodeManagers, available memory, ...). */
String getClusterDescription();

/* 查询已存在的 Flink 集群. */
ClusterClientProvider&lt;T&gt; retrieve(T clusterId) throws ClusterRetrieveException;

/** 创建 Flink Session 集群 */
ClusterClientProvider&lt;T&gt; deploySessionCluster(ClusterSpecification clusterSpecification)
        throws ClusterDeploymentException;

/** 创建 Flink Application 集群 **/
ClusterClientProvider&lt;T&gt; deployApplicationCluster(
        final ClusterSpecification clusterSpecification,
        final ApplicationConfiguration applicationConfiguration)
        throws ClusterDeploymentException;

/** 创建 Per-job 集群 **/
ClusterClientProvider&lt;T&gt; deployJobCluster(
        final ClusterSpecification clusterSpecification,
        final JobGraph jobGraph,
        final boolean detached)
        throws ClusterDeploymentException;

/** 删除集群 **/
void killCluster(T clusterId) throws FlinkException;

@Override
void close();

}

public class KubernetesClusterDescriptor implements ClusterDescriptor<String> {

private final Configuration flinkConfig;

	// 内置 k8s client
private final FlinkKubeClient client;

private final String clusterId;

@Override
public ClusterClientProvider&lt;String&gt; deployApplicationCluster(
        final ClusterSpecification clusterSpecification,
        final ApplicationConfiguration applicationConfiguration)
        throws ClusterDeploymentException {
  	// 查询 flink 集群在 k8s 中是否存在
    if (client.getRestService(clusterId).isPresent()) {
        throw new ClusterDeploymentException(
                &#34;The Flink cluster &#34; + clusterId + &#34; already exists.&#34;);
    }

    final KubernetesDeploymentTarget deploymentTarget =
            KubernetesDeploymentTarget.fromConfig(flinkConfig);
    if (KubernetesDeploymentTarget.APPLICATION != deploymentTarget) {
        throw new ClusterDeploymentException(
                &#34;Couldn&#39;t deploy Kubernetes Application Cluster.&#34;
                        + &#34; Expected deployment.target=&#34;
                        + KubernetesDeploymentTarget.APPLICATION.getName()
                        + &#34; but actual one was \&#34;&#34;
                        + deploymentTarget
                        + &#34;\&#34;&#34;);
    }
			
  	// 设置 application 参数：$internal.application.program-args 和 $internal.application.main
    applicationConfiguration.applyToConfiguration(flinkConfig);
  
    // 创建集群
    final ClusterClientProvider&lt;String&gt; clusterClientProvider =
            deployClusterInternal(
                    KubernetesApplicationClusterEntrypoint.class.getName(),
                    clusterSpecification,
                    false);

    try (ClusterClient&lt;String&gt; clusterClient = clusterClientProvider.getClusterClient()) {
        LOG.info(
                &#34;Create flink application cluster {} successfully, JobManager Web Interface: {}&#34;,
                clusterId,
                clusterClient.getWebInterfaceURL());
    }
    return clusterClientProvider;
}

// 创建集群逻辑
private ClusterClientProvider&lt;String&gt; deployClusterInternal(
        String entryPoint, ClusterSpecification clusterSpecification, boolean detached)
        throws ClusterDeploymentException {
    final ClusterEntrypoint.ExecutionMode executionMode =
            detached
                    ? ClusterEntrypoint.ExecutionMode.DETACHED
                    : ClusterEntrypoint.ExecutionMode.NORMAL;
    flinkConfig.setString(
            ClusterEntrypoint.INTERNAL_CLUSTER_EXECUTION_MODE, executionMode.toString());

    flinkConfig.setString(KubernetesConfigOptionsInternal.ENTRY_POINT_CLASS, entryPoint);

    // Rpc, blob, rest, taskManagerRpc ports need to be exposed, so update them to fixed values.
  	// 将端口指定为固定值，方便 k8s 的资源构建。因为 pod 的隔离性，所以没有端口冲突
    KubernetesUtils.checkAndUpdatePortConfigOption(
            flinkConfig, BlobServerOptions.PORT, Constants.BLOB_SERVER_PORT);
    KubernetesUtils.checkAndUpdatePortConfigOption(
            flinkConfig, TaskManagerOptions.RPC_PORT, Constants.TASK_MANAGER_RPC_PORT);
    KubernetesUtils.checkAndUpdatePortConfigOption(
            flinkConfig, RestOptions.BIND_PORT, Constants.REST_PORT);
			// HA 配置
    if (HighAvailabilityMode.isHighAvailabilityModeActivated(flinkConfig)) {
        flinkConfig.setString(HighAvailabilityOptions.HA_CLUSTER_ID, clusterId);
        KubernetesUtils.checkAndUpdatePortConfigOption(
                flinkConfig,
                HighAvailabilityOptions.HA_JOB_MANAGER_PORT_RANGE,
                flinkConfig.get(JobManagerOptions.PORT));
    }

    try {
        final KubernetesJobManagerParameters kubernetesJobManagerParameters =
                new KubernetesJobManagerParameters(flinkConfig, clusterSpecification);
					// 补充 PodTemplate 逻辑
        final FlinkPod podTemplate =
                kubernetesJobManagerParameters
                        .getPodTemplateFilePath()
                        .map(
                                file -&gt;
                                        KubernetesUtils.loadPodFromTemplateFile(
                                                client, file, Constants.MAIN_CONTAINER_NAME))
                        .orElse(new FlinkPod.Builder().build());
        final KubernetesJobManagerSpecification kubernetesJobManagerSpec =
                KubernetesJobManagerFactory.buildKubernetesJobManagerSpecification(
                        podTemplate, kubernetesJobManagerParameters);
      
					// 核心逻辑：在 k8s 中创建包括 JobManager Deployment 在内 k8s 资源，比如 Service 和 ConfigMap
        client.createJobManagerComponent(kubernetesJobManagerSpec);

        return createClusterClientProvider(clusterId);
    } catch (Exception e) {
        //...
    }
}

}