Flink 内存管理

// org.apache.flink.yarn.YarnClusterDescriptor
// 提交任务中在 clusterDescriptor.deployJobCluster 调用
private ClusterSpecification validateClusterResources(
  ClusterSpecification clusterSpecification,
  int yarnMinAllocationMB,
  Resource maximumResourceCapability,
  ClusterResourceDescription freeClusterResources) throws YarnDeploymentException {
  // JM 内存默认 768M，TM 内存默认1G
  int jobManagerMemoryMb = clusterSpecification.getMasterMemoryMB();
  final int taskManagerMemoryMb = clusterSpecification.getTaskManagerMemoryMB();
  logIfComponentMemNotIntegerMultipleOfYarnMinAllocation("JobManager", jobManagerMemoryMb, yarnMinAllocationMB);
  logIfComponentMemNotIntegerMultipleOfYarnMinAllocation("TaskManager", taskManagerMemoryMb, yarnMinAllocationMB);
  // set the memory to minAllocationMB to do the next checks correctly
  if (jobManagerMemoryMb < yarnMinAllocationMB) {
    jobManagerMemoryMb =  yarnMinAllocationMB;
  }
  final String note = "Please check the 'yarn.scheduler.maximum-allocation-mb' and the 'yarn.nodemanager.resource.memory-mb' configuration values\n";
  if (jobManagerMemoryMb > maximumResourceCapability.getMemory()) {
    throw new YarnDeploymentException("The cluster does not have the requested resources for the JobManager available!\n"
        + "Maximum Memory: " + maximumResourceCapability.getMemory() + "MB Requested: " + jobManagerMemoryMb + "MB. " + note);
  }
  if (taskManagerMemoryMb > maximumResourceCapability.getMemory()) {
    throw new YarnDeploymentException("The cluster does not have the requested resources for the TaskManagers available!\n"
        + "Maximum Memory: " + maximumResourceCapability.getMemory() + " Requested: " + taskManagerMemoryMb + "MB. " + note);
  }
  // 集群内存资源是否充足
  final String noteRsc = "\nThe Flink YARN client will try to allocate the YARN session, but maybe not all TaskManagers are " +
      "connecting from the beginning because the resources are currently not available in the cluster. " +
      "The allocation might take more time than usual because the Flink YARN client needs to wait until " +
      "the resources become available.";
  if (taskManagerMemoryMb > freeClusterResources.containerLimit) {
    LOG.warn("The requested amount of memory for the TaskManagers (" + taskManagerMemoryMb + "MB) is more than "
        + "the largest possible YARN container: " + freeClusterResources.containerLimit + noteRsc);
  }
  if (jobManagerMemoryMb > freeClusterResources.containerLimit) {
    LOG.warn("The requested amount of memory for the JobManager (" + jobManagerMemoryMb + "MB) is more than "
        + "the largest possible YARN container: " + freeClusterResources.containerLimit + noteRsc);
  }
  return new ClusterSpecification.ClusterSpecificationBuilder()
      .setMasterMemoryMB(jobManagerMemoryMb)
      .setTaskManagerMemoryMB(taskManagerMemoryMb)
      .setSlotsPerTaskManager(clusterSpecification.getSlotsPerTaskManager())
      .createClusterSpecification();

}

// TM 内存分配，提交任务中生成ClusterSpecification 时调用
// org.apache.flink.runtime.clusterframework.TaskExecutorProcessUtils
public static TaskExecutorProcessSpec processSpecFromConfig(final Configuration config) {
	return createMemoryProcessSpec(config, PROCESS_MEMORY_UTILS.memoryProcessSpecFromConfig(config));
}
private static TaskExecutorProcessSpec createMemoryProcessSpec(
		final Configuration config,
		final CommonProcessMemorySpec<TaskExecutorFlinkMemory> processMemory) {
  // CommonProcessMemorySpec TaskExecutorProcessSpec 类注释描述内存分布
	TaskExecutorFlinkMemory flinkMemory = processMemory.getFlinkMemory();
	JvmMetaspaceAndOverhead jvmMetaspaceAndOverhead = processMemory.getJvmMetaspaceAndOverhead();
	// Process = Flink + jvm meta + jvm overhead
  return new TaskExecutorProcessSpec(getCpuCores(config), flinkMemory, jvmMetaspaceAndOverhead);
}

// org.apache.flink.runtime.util.config.memory.ProcessMemoryUtils
public CommonProcessMemorySpec<FM> memoryProcessSpecFromConfig(Configuration config) {
	if (options.getRequiredFineGrainedOptions().stream().allMatch(config::contains)) {
		// all internal memory options are configured, use these to derive total Flink and process memory
		return deriveProcessSpecWithExplicitInternalMemory(config);
	} else if (config.contains(options.getTotalFlinkMemoryOption())) {
		// internal memory options are not configured, total Flink memory is configured,
		// derive from total flink memory
		return deriveProcessSpecWithTotalFlinkMemory(config);
	} else if (config.contains(options.getTotalProcessMemoryOption())) {
		// total Flink memory is not configured, total process memory is configured,
		// derive from total process memory
    // 设置 TM 总内存为基点，开始计算每个部件占用内存
		return deriveProcessSpecWithTotalProcessMemory(config);
	}
	return failBecauseRequiredOptionsNotConfigured();
}
private CommonProcessMemorySpec<FM> deriveProcessSpecWithTotalProcessMemory(Configuration config) {
	MemorySize totalProcessMemorySize = getMemorySizeFromConfig(config, options.getTotalProcessMemoryOption());
	// 根据比例得到 JvmMetaspaceAndOverhead
  JvmMetaspaceAndOverhead jvmMetaspaceAndOverhead =
		deriveJvmMetaspaceAndOverheadWithTotalProcessMemory(config, totalProcessMemorySize);
	// totalFlinkMemorySize = totalProcessMemorySize - jvmMetaspaceAndOverhead
  MemorySize totalFlinkMemorySize = totalProcessMemorySize.subtract(
		jvmMetaspaceAndOverhead.getTotalJvmMetaspaceAndOverheadSize());
	FM flinkInternalMemory = flinkMemoryUtils.deriveFromTotalFlinkMemory(config, totalFlinkMemorySize);
	return new CommonProcessMemorySpec<>(flinkInternalMemory, jvmMetaspaceAndOverhead);
}

// org.apache.flink.runtime.util.config.memory.taskmanager.TaskExecutorFlinkMemoryUtils
// 计算 Flink 内存分布
public TaskExecutorFlinkMemory deriveFromTotalFlinkMemory(
	final Configuration config,
	final MemorySize totalFlinkMemorySize) {
  // 获取 flink framework 和task 对外内存大小
	final MemorySize frameworkHeapMemorySize = getFrameworkHeapMemorySize(config);
	final MemorySize frameworkOffHeapMemorySize = getFrameworkOffHeapMemorySize(config);
	final MemorySize taskOffHeapMemorySize = getTaskOffHeapMemorySize(config);

	final MemorySize taskHeapMemorySize;
	final MemorySize networkMemorySize;
	final MemorySize managedMemorySize;
  // flink 根据task 内存是否显式分配，划分为两种方式：
	if (isTaskHeapMemorySizeExplicitlyConfigured(config)) {
    // task 直接分配时，network = Flink - framework - task - managed
		taskHeapMemorySize = getTaskHeapMemorySize(config);
		managedMemorySize = deriveManagedMemoryAbsoluteOrWithFraction(config, totalFlinkMemorySize);
		final MemorySize totalFlinkExcludeNetworkMemorySize =
			frameworkHeapMemorySize.add(frameworkOffHeapMemorySize).add(taskHeapMemorySize).add(taskOffHeapMemorySize).add(managedMemorySize);
		if (totalFlinkExcludeNetworkMemorySize.getBytes() > totalFlinkMemorySize.getBytes()) {
			...
		}
		networkMemorySize = totalFlinkMemorySize.subtract(totalFlinkExcludeNetworkMemorySize);
		sanityCheckNetworkMemoryWithExplicitlySetTotalFlinkAndHeapMemory(config, networkMemorySize, totalFlinkMemorySize);
	} else {
    // 未直接设置 task，task = Flink - framework - task offHeap - managed - network
		managedMemorySize = deriveManagedMemoryAbsoluteOrWithFraction(config, totalFlinkMemorySize);

		networkMemorySize = isUsingLegacyNetworkConfigs(config) ? getNetworkMemorySizeWithLegacyConfig(config) :
			deriveNetworkMemoryWithFraction(config, totalFlinkMemorySize);
		final MemorySize totalFlinkExcludeTaskHeapMemorySize =
			frameworkHeapMemorySize.add(frameworkOffHeapMemorySize).add(taskOffHeapMemorySize).add(managedMemorySize).add(networkMemorySize);
		if (totalFlinkExcludeTaskHeapMemorySize.getBytes() > totalFlinkMemorySize.getBytes()) {
			...
		}
		taskHeapMemorySize = totalFlinkMemorySize.subtract(totalFlinkExcludeTaskHeapMemorySize);
	}
	final TaskExecutorFlinkMemory flinkInternalMemory = new TaskExecutorFlinkMemory(
		frameworkHeapMemorySize,
		frameworkOffHeapMemorySize,
		taskHeapMemorySize,
		taskOffHeapMemorySize,
		networkMemorySize,
		managedMemorySize);
	sanityCheckTotalFlinkMemory(config, flinkInternalMemory);

	return flinkInternalMemory;
}

// org.apache.flink.runtime.taskexecutor.TaskExecutorResourceUtils
// org.apache.flink.configuration.TaskManagerOptions
public static final ConfigOption<MemorySize> NETWORK_MEMORY_MIN =
  key("taskmanager.memory.network.min")
  .memoryType()
  .defaultValue(MemorySize.parse("64m"))
public static final ConfigOption<MemorySize> NETWORK_MEMORY_MAX =
  key("taskmanager.memory.network.max")
    .memoryType()
    .defaultValue(MemorySize.parse("1g"))
public static final ConfigOption<Float> NETWORK_MEMORY_FRACTION =
  key("taskmanager.memory.network.fraction")
    .floatType()
    .defaultValue(0.1f)
private static void adjustNetworkMemoryForLocalExecution(Configuration config) {
  if (!config.contains(TaskManagerOptions.NETWORK_MEMORY_MIN) &&
    config.contains(TaskManagerOptions.NETWORK_MEMORY_MAX)) {
    config.set(TaskManagerOptions.NETWORK_MEMORY_MIN, config.get(TaskManagerOptions.NETWORK_MEMORY_MAX));
  }
  if (!config.contains(TaskManagerOptions.NETWORK_MEMORY_MAX) &&
    config.contains(TaskManagerOptions.NETWORK_MEMORY_MIN)) {
    config.set(TaskManagerOptions.NETWORK_MEMORY_MAX, config.get(TaskManagerOptions.NETWORK_MEMORY_MIN));
  }
  setConfigOptionToDefaultIfNotSet(config, TaskManagerOptions.NETWORK_MEMORY_MIN, DEFAULT_SHUFFLE_MEMORY_SIZE);
  setConfigOptionToDefaultIfNotSet(config, TaskManagerOptions.NETWORK_MEMORY_MAX, DEFAULT_SHUFFLE_MEMORY_SIZE);
}