flink-ui中的反压采样

发表于 | 分类于 | | |
字数统计 1.1k | 阅读时长 5

flink反压值采样计算原理

我们在点击flink ui的operator->backpressure之后，会触发Backpressure采样：每隔BACK_PRESSURE_REFRESH_INTERVAL的间隔进行一次采样。

BackPressureStatsTracker#triggerStackTraceSample

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
 /**
 * Triggers a stack trace sample for a operator to gather the back pressure
 * statistics. If there is a sample in progress for the operator, the call
 * is ignored.
 *
 * @param vertex Operator to get the stats for. 代表要采样的Operator节点
 * @return Flag indicating whether a sample with triggered.
 */
@SuppressWarnings("unchecked")
public boolean triggerStackTraceSample(ExecutionJobVertex vertex) {
	// 保证没有并行triggerSimple
	synchronized (lock) {
		if (shutDown) {
			return false;
		}
		//排除掉已经pending在采样和结束的Operator
		if (!pendingStats.contains(vertex) &&
				!vertex.getGraph().getState().isGloballyTerminalState()) {
				
			// 拿到对应ExecutionGraph的FutureExecutor，这是用以在相应的ExecutionGraph发起任务的
			Executor executor = vertex.getGraph().getFutureExecutor();

			// Only trigger if still active job
			if (executor != null) {
				pendingStats.add(vertex);

				if (LOG.isDebugEnabled()) {
					LOG.debug("Triggering stack trace sample for tasks: " + Arrays.toString(vertex.getTaskVertices()));
				}
				// 核心方法 通过StackTraceSampleCoordinator 去发起相应的trigger流程, 这里的Future是Flink内部自己定义的异步结果接口 具体可查看FlinkFuture.java(可以深入了解下)
				Future<StackTraceSample> sample = coordinator.triggerStackTraceSample(
						vertex.getTaskVertices(),
						numSamples,
						delayBetweenSamples,
						MAX_STACK_TRACE_DEPTH);
				// 指定异步回调函数（采样结果分析的函数）
				sample.handleAsync(new StackTraceSampleCompletionCallback(vertex), executor);

				return true;
			}
		}

		return false;
	}
}

StackTraceSampleCoordinator#triggerStackTraceSample

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
/**
	 * Triggers a stack trace sample to all tasks.
	 *
	 * @param tasksToSample       Tasks to sample.
	 * @param numSamples          Number of stack trace samples to collect.
	 * @param delayBetweenSamples Delay between consecutive samples.
	 * @param maxStackTraceDepth  Maximum depth of the stack trace. 0 indicates
	 *                            no maximum and keeps the complete stack trace.
	 * @return A future of the completed stack trace sample
	 */
	@SuppressWarnings("unchecked")
	public Future<StackTraceSample> triggerStackTraceSample(
			ExecutionVertex[] tasksToSample,
			int numSamples,
			Time delayBetweenSamples,
			int maxStackTraceDepth) {

		checkNotNull(tasksToSample, "Tasks to sample");
		checkArgument(tasksToSample.length >= 1, "No tasks to sample");
		checkArgument(numSamples >= 1, "No number of samples");
		checkArgument(maxStackTraceDepth >= 0, "Negative maximum stack trace depth");

		// 通过ExecutionVertex获取ExecutionAttemptID和Execution，并最后做存活判断
		// Execution IDs of running tasks
		ExecutionAttemptID[] triggerIds = new ExecutionAttemptID[tasksToSample.length];
		Execution[] executions = new Execution[tasksToSample.length];

		// Check that all tasks are RUNNING before triggering anything. The
		// triggering can still fail.
		for (int i = 0; i < triggerIds.length; i++) {
			Execution execution = tasksToSample[i].getCurrentExecutionAttempt();
			if (execution != null && execution.getState() == ExecutionState.RUNNING) {
				executions[i] = execution;
				triggerIds[i] = execution.getAttemptId();
			} else {
				return FlinkCompletableFuture.completedExceptionally(
					new IllegalStateException("Task " + tasksToSample[i]
					.getTaskNameWithSubtaskIndex() + " is not running."));
			}
		}

		synchronized (lock) {
			if (isShutDown) {
				return FlinkCompletableFuture.completedExceptionally(new IllegalStateException("Shut down"));
			}

			final int sampleId = sampleIdCounter++;

			LOG.debug("Triggering stack trace sample {}", sampleId);
			
			// 包含采样id和ExecutionAttemptID
			final PendingStackTraceSample pending = new PendingStackTraceSample(
					sampleId, triggerIds);

			// Discard the sample if it takes too long. We don't send cancel
			// messages to the task managers, but only wait for the responses
			// and then ignore them.
			long expectedDuration = numSamples * delayBetweenSamples.toMilliseconds();
			Time timeout = Time.milliseconds(expectedDuration + sampleTimeout);

			// Add the pending sample before scheduling the discard task to
			// prevent races with removing it again.
			pendingSamples.put(sampleId, pending);

			// Trigger all samples
			// execution是executionVertex的多次执行（recovery...）
			for (Execution execution: executions) {
			    // 对相应的execution进行多次numSamples采样，但是都是同一个sampleId
				final Future<StackTraceSampleResponse> stackTraceSampleFuture = execution.requestStackTraceSample(
					sampleId,
					numSamples,
					delayBetweenSamples,
					maxStackTraceDepth,
					timeout);

				stackTraceSampleFuture.handleAsync(new BiFunction<StackTraceSampleResponse, Throwable, Void>() {
					@Override
					public Void apply(StackTraceSampleResponse stackTraceSampleResponse, Throwable throwable) {
						if (stackTraceSampleResponse != null) {
						    // 收集返回的List<StackTraceElement[]> 到PendingStackTraceSample
							collectStackTraces(
								stackTraceSampleResponse.getSampleId(),
								stackTraceSampleResponse.getExecutionAttemptID(),
								// 返回的堆栈信息包含所有的采样结果
								stackTraceSampleResponse.getSamples());
						} else {
							cancelStackTraceSample(sampleId, throwable);
						}

						return null;
					}
				}, executor);
			}

			return pending.getStackTraceSampleFuture();
		}
	}

BackPressureStatsTracker#StackTraceSampleCompletionCallback

采样的结果就是StackTraceSample

1
2
3
4
5
* java.lang.Object.wait(Native Method)
* o.a.f.[...].LocalBufferPool.requestBuffer(LocalBufferPool.java:163)
* o.a.f.[...].LocalBufferPool.requestBufferBlocking(LocalBufferPool.java:133) <--- BLOCKING
* request
* [...]

利用这样的线程堆栈类型来判断是否block住了。

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
/**
		 * Creates the back pressure stats from a stack trace sample.
		 *
		 * @param sample Stack trace sample to base stats on.
		 *
		 * @return Back pressure stats
		 */
		private OperatorBackPressureStats createStatsFromSample(StackTraceSample sample) {
			Map<ExecutionAttemptID, List<StackTraceElement[]>> traces = sample.getStackTraces();

			// Map task ID to subtask index, because the web interface expects
			// it like that.
			// 方便下面根据executionId查询相应的并发度
			Map<ExecutionAttemptID, Integer> subtaskIndexMap = Maps
					.newHashMapWithExpectedSize(traces.size());

			Set<ExecutionAttemptID> sampledTasks = sample.getStackTraces().keySet();

			for (ExecutionVertex task : vertex.getTaskVertices()) {
				ExecutionAttemptID taskId = task.getCurrentExecutionAttempt().getAttemptId();
				if (sampledTasks.contains(taskId)) {
					subtaskIndexMap.put(taskId, task.getParallelSubtaskIndex());
				} else {
					LOG.debug("Outdated sample. A task, which is part of the " +
							"sample has been reset.");
				}
			}

			// Ratio of blocked samples to total samples per sub task. Array
			// position corresponds to sub task index.
			// 数组的index和task的并发度相绑定
			double[] backPressureRatio = new double[traces.size()];

			for (Entry<ExecutionAttemptID, List<StackTraceElement[]>> entry : traces.entrySet()) {
				int backPressureSamples = 0;

				List<StackTraceElement[]> taskTraces = entry.getValue();

				for (StackTraceElement[] trace : taskTraces) {
					for (int i = trace.length - 1; i >= 0; i--) {
						StackTraceElement elem = trace[i];

						if (elem.getClassName().equals(EXPECTED_CLASS_NAME) &&
								elem.getMethodName().equals(EXPECTED_METHOD_NAME)) {

							backPressureSamples++;
							break; // Continue with next stack trace
						}
					}
				}

				int subtaskIndex = subtaskIndexMap.get(entry.getKey());

				int size = taskTraces.size();
				double ratio = (size > 0)
						? ((double) backPressureSamples) / size
						: 0;

				backPressureRatio[subtaskIndex] = ratio;
			}

			return new OperatorBackPressureStats(
					sample.getSampleId(),
					sample.getEndTime(),
					backPressureRatio);
		}
	}

至此完成采样

待学习

  1. Flink反压原理
  2. 理解清里面很多的Future使用方法
谢谢支持