flink基础教程

val kafkaSource = new FlinkKafkaConsumer[String]("kafka_offset", new SimpleStringSchema(), prop)
val kafkaSource1 = new FlinkKafkaConsumer[String]("kafka_offset", new SimpleStringSchema(), prop)
val kafkaProducer = new FlinkKafkaProducer[String]("kafka_offset_out", new SimpleStringSchema(), prop)
val source = env
  .addSource(kafkaSource)
  .setParallelism(1)
  .name("source")
val source1 = env
  .addSource(kafkaSource1)
  .setParallelism(1)
  .name("source1")

source.union(source1)
  .map(node => {
    node.toString + ",flinkx"
  })
  .addSink(kafkaProducer)

private final class ScheduledTrigger implements Runnable {

    @Override
    public void run() {
        try {
            triggerCheckpoint(true);
        }
        catch (Exception e) {
            LOG.error("Exception while triggering checkpoint for job {}.", job, e);
        }
    }
}

private void startScheduling() {
        checkState(jobStatusListener == null);
        // register self as job status change listener
        jobStatusListener = new JobManagerJobStatusListener();
        schedulerNG.registerJobStatusListener(jobStatusListener);
        // 这个会调用到 CheckpointCoordinator.scheduleTriggerWithDelay 方法启动第一次 Checkpoint，后续就由 CheckpointCoordinator 自己启动
        schedulerNG.startScheduling();
    }

private ScheduledFuture<?> scheduleTriggerWithDelay(long initDelay) {
    return timer.scheduleAtFixedRate(
        new ScheduledTrigger(),
        initDelay, baseInterval, TimeUnit.MILLISECONDS);
}

/**
 * Executes the given command periodically. The first execution is started after the
 * {@code initialDelay}, the second execution is started after {@code initialDelay + period},
 * the third after {@code initialDelay + 2*period} and so on.
 * The task is executed until either an execution fails, or the returned {@link ScheduledFuture}
 * is cancelled.
 *
 * @param command the task to be executed periodically
 * @param initialDelay the time from now until the first execution is triggered  第一次启动 Checkpoint 时间
 * @param period the time after which the next execution is triggered            后续的时间间隔
 * @param unit the time unit of the delay and period parameter
 * @return a ScheduledFuture representing the periodic task. This future never completes
 * unless an execution of the given task fails or if the future is cancelled
 */
ScheduledFuture<?> scheduleAtFixedRate(
    Runnable command,
    long initialDelay,
    long period,
    TimeUnit unit);

public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(boolean isPeriodic) {
    return triggerCheckpoint(checkpointProperties, null, isPeriodic, false);
}

@VisibleForTesting
public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(
        CheckpointProperties props,
        @Nullable String externalSavepointLocation,
        boolean isPeriodic,
        boolean advanceToEndOfTime) {


    CheckpointTriggerRequest request = new CheckpointTriggerRequest(props, externalSavepointLocation, isPeriodic, advanceToEndOfTime);
    requestDecider
        .chooseRequestToExecute(request, isTriggering, lastCheckpointCompletionRelativeTime)
        //  调用 startTriggeringCheckpoint 方法
        .ifPresent(this::startTriggeringCheckpoint);
    return request.onCompletionPromise;
}

private void startTriggeringCheckpoint(CheckpointTriggerRequest request) {

    // 获取 需要触发 Checkpoint 的算子
    final Execution[] executions = getTriggerExecutions();
....
    // no exception, no discarding, everything is OK
    final long checkpointId = checkpoint.getCheckpointId();
    snapshotTaskState(
        timestamp,
        checkpointId,
        checkpoint.getCheckpointStorageLocation(),
        request.props,
        executions,
        request.advanceToEndOfTime);

    coordinatorsToCheckpoint.forEach((ctx) -> ctx.afterSourceBarrierInjection(checkpointId));

    onTriggerSuccess();
............
}


private void snapshotTaskState(
        long timestamp,
        long checkpointID,
        CheckpointStorageLocation checkpointStorageLocation,
        CheckpointProperties props,
        Execution[] executions,
        boolean advanceToEndOfTime) {

        // send the messages to the tasks that trigger their checkpoint
        // 给每个 Execution (在这里可以理解为每个 Source 算子，因为Checkpoint 是从 Source 开始的) 发送 trigger checkpoint 消息
        for (Execution execution: executions) {
            if (props.isSynchronous()) {
                execution.triggerSynchronousSavepoint(checkpointID, timestamp, checkpointOptions, advanceToEndOfTime);
            } else {
                execution.triggerCheckpoint(checkpointID, timestamp, checkpointOptions);
            }
        }
    }

private void triggerCheckpointHelper(long checkpointId, long timestamp, CheckpointOptions checkpointOptions, boolean advanceToEndOfEventTime) {

    final CheckpointType checkpointType = checkpointOptions.getCheckpointType();
    if (advanceToEndOfEventTime && !(checkpointType.isSynchronous() && checkpointType.isSavepoint())) {
        throw new IllegalArgumentException("Only synchronous savepoints are allowed to advance the watermark to MAX.");
    }

    final LogicalSlot slot = assignedResource;

    if (slot != null) {
        final TaskManagerGateway taskManagerGateway = slot.getTaskManagerGateway();

        taskManagerGateway.triggerCheckpoint(attemptId, getVertex().getJobId(), checkpointId, timestamp, checkpointOptions, advanceToEndOfEventTime);
    } else {
        LOG.debug("The execution has no slot assigned. This indicates that the execution is no longer running.");
    }
}

@Override
public CompletableFuture<Acknowledge> triggerCheckpoint(
        ExecutionAttemptID executionAttemptID,
        long checkpointId,
        long checkpointTimestamp,
        CheckpointOptions checkpointOptions,
        boolean advanceToEndOfEventTime) {
    log.debug("Trigger checkpoint {}@{} for {}.", checkpointId, checkpointTimestamp, executionAttemptID);

    final CheckpointType checkpointType = checkpointOptions.getCheckpointType();
    if (advanceToEndOfEventTime && !(checkpointType.isSynchronous() && checkpointType.isSavepoint())) {
        throw new IllegalArgumentException("Only synchronous savepoints are allowed to advance the watermark to MAX.");
    }
    // 使用 operator id 获取 job 对应的 Source 算子 （如果有多个 Source，JobManager 端会发送两个 Rpc 请求，TaskManager 也是执行两次）
    final Task task = taskSlotTable.getTask(executionAttemptID);

    if (task != null) {
        // 对 task 触发 一个 Checkpoint Barrier
        task.triggerCheckpointBarrier(checkpointId, checkpointTimestamp, checkpointOptions, advanceToEndOfEventTime);

        return CompletableFuture.completedFuture(Acknowledge.get());
    } else {
        final String message = "TaskManager received a checkpoint request for unknown task " + executionAttemptID + '.';

        log.debug(message);
        return FutureUtils.completedExceptionally(new CheckpointException(message, CheckpointFailureReason.TASK_CHECKPOINT_FAILURE));
    }
}

TaskExecutor.triggerCheckpoint  --> task.triggerCheckpointBarrier

Task.triggerCheckpointBarrier   --> invokable.triggerCheckpointAsync

SourceStreamTask.triggerCheckpointAsync   -->   super.triggerCheckpointAsync

StreamTask.triggerCheckpointAsync   -->  triggerCheckpoint

StreamTask.triggerCheckpoint  ---> performCheckpoint

StreamTask.performCheckpoint  --> subtaskCheckpointCoordinator.checkpointState

SubtaskCheckpointCoordinatorImpl.checkpointState       --->  takeSnapshotSync    (会广播 CheckpointBarrier 到下游算子, takeSnapshotSync 成成功后会发送 ack 到 JobMaster)

SubtaskCheckpointCoordinatorImpl.takeSnapshotSync   --->  buildOperatorSnapshotFutures

SubtaskCheckpointCoordinatorImpl.buildOperatorSnapshotFutures   ---> checkpointStreamOperator

SubtaskCheckpointCoordinatorImpl.checkpointStreamOperator   --->  op.snapshotState

AbstractStreamOperator.snapshotState   ---> stateHandler.snapshotState

StreamOperatorStateHandler.snapshotState   --->  snapshotState              (会调用 operatorStateBackend.snapshot/keyedStateBackend.snapshot 将 state 持久化到 stateBackend )

StreamOperatorStateHandler.snapshotState   --->  streamOperator.snapshotState(snapshotContext)

AbstractUdfStreamOperator.snapshotState   ---> StreamingFunctionUtils.snapshotFunctionState

StreamingFunctionUtils.snapshotFunctionState   ---> trySnapshotFunctionState

StreamingFunctionUtils.trySnapshotFunctionState   ---> ((CheckpointedFunction) userFunction).snapshotState  (就进入 udf 了，这里是：FlinkKafkaConsumerBase， 如果是自定义的 Source，会进入对应Source 的 snapshotState 方法)

FlinkKafkaConsumerBase.snapshotState

/** Accessor for state in the operator state backend. */
// ListState 存储状态，Checkpoint 的时候只需要将 KafkaTopicPartition 和 Offset 放入这个对象中，Checkpoint 的时候，就会写入 statebackend
// Operator state 都是这样的，自己实现的也是，将对应内容写入状态就可以了
private transient ListState<Tuple2<KafkaTopicPartition, Long>> unionOffsetStates;


@Override
public final void snapshotState(FunctionSnapshotContext context) throws Exception {
    if (!running) {
        LOG.debug("snapshotState() called on closed source");
    } else {
        // 消费者还在运行
        // 清楚之前的状态
        unionOffsetStates.clear();

        final AbstractFetcher<?, ?> fetcher = this.kafkaFetcher;
        if (fetcher == null) {
            // the fetcher has not yet been initialized, which means we need to return the
            // originally restored offsets or the assigned partitions
            for (Map.Entry<KafkaTopicPartition, Long> subscribedPartition : subscribedPartitionsToStartOffsets.entrySet()) {
                unionOffsetStates.add(Tuple2.of(subscribedPartition.getKey(), subscribedPartition.getValue()));
            }

            if (offsetCommitMode == OffsetCommitMode.ON_CHECKPOINTS) {
                // the map cannot be asynchronously updated, because only one checkpoint call can happen
                // on this function at a time: either snapshotState() or notifyCheckpointComplete()
                pendingOffsetsToCommit.put(context.getCheckpointId(), restoredState);
            }
        } else {
            // 获取对应的 KafkaTopicPartition 和 offset
            HashMap<KafkaTopicPartition, Long> currentOffsets = fetcher.snapshotCurrentState();
            // 放到 pendingOffsetsToCommit 中
            if (offsetCommitMode == OffsetCommitMode.ON_CHECKPOINTS) {
                // the map cannot be asynchronously updated, because only one checkpoint call can happen
                // on this function at a time: either snapshotState() or notifyCheckpointComplete()
                pendingOffsetsToCommit.put(context.getCheckpointId(), currentOffsets);
            }
            // 将 KafkaTopicPartition 和 offset 写入 operator state 中
            for (Map.Entry<KafkaTopicPartition, Long> kafkaTopicPartitionLongEntry : currentOffsets.entrySet()) {
                unionOffsetStates.add(
                        Tuple2.of(kafkaTopicPartitionLongEntry.getKey(), kafkaTopicPartitionLongEntry.getValue()));
            }
        }
        // 移除还未 提交的 offset
        if (offsetCommitMode == OffsetCommitMode.ON_CHECKPOINTS) {
            // truncate the map of pending offsets to commit, to prevent infinite growth
            while (pendingOffsetsToCommit.size() > MAX_NUM_PENDING_CHECKPOINTS) {
                pendingOffsetsToCommit.remove(0);
            }
        }
    }
}

fetcher.commitInternalOffsetsToKafka(offsets, offsetCommitCallback);

private void finishAndReportAsync(Map<OperatorID, OperatorSnapshotFutures> snapshotFutures, CheckpointMetaData metadata, CheckpointMetrics metrics, CheckpointOptions options) {
    // we are transferring ownership over snapshotInProgressList for cleanup to the thread, active on submit
    // AsyncCheckpointRunnable 的 run 方法，会给 JobMaster 发送 Checkpoint 完成的消息
    executorService.execute(new AsyncCheckpointRunnable(
        snapshotFutures,
        metadata,
        metrics,
        System.nanoTime(),
        taskName,
        registerConsumer(),
        unregisterConsumer(),
        env,
        asyncExceptionHandler));
}

public void run() {

    ..........
    if (asyncCheckpointState.compareAndSet(AsyncCheckpointState.RUNNING, AsyncCheckpointState.COMPLETED)) {
        // report ack
        reportCompletedSnapshotStates(
            jobManagerTaskOperatorSubtaskStates,
            localTaskOperatorSubtaskStates,
            asyncDurationMillis);

    } else {
        LOG.debug("{} - asynchronous part of checkpoint {} could not be completed because it was closed before.",
            taskName,
            checkpointMetaData.getCheckpointId());
    }
    .........

private void reportCompletedSnapshotStates(
    TaskStateSnapshot acknowledgedTaskStateSnapshot,
    TaskStateSnapshot localTaskStateSnapshot,
    long asyncDurationMillis) {


    // we signal stateless tasks by reporting null, so that there are no attempts to assign empty state
    // to stateless tasks on restore. This enables simple job modifications that only concern
    // stateless without the need to assign them uids to match their (always empty) states.
    taskEnvironment.getTaskStateManager().reportTaskStateSnapshots(
        checkpointMetaData,
        checkpointMetrics,
        hasAckState ? acknowledgedTaskStateSnapshot : null,
        hasLocalState ? localTaskStateSnapshot : null);

}

@Override
public InputStatus emitNext(DataOutput<T> output) throws Exception {

    while (true) {
        // get the stream element from the deserializer
        if (currentRecordDeserializer != null) {
            DeserializationResult result = currentRecordDeserializer.getNextRecord(deserializationDelegate);
            if (result.isBufferConsumed()) {
                currentRecordDeserializer.getCurrentBuffer().recycleBuffer();
                currentRecordDeserializer = null;
            }

            if (result.isFullRecord()) {
                // 数据的处理流程，从这里调用直到 user function 的map 中
                processElement(deserializationDelegate.getInstance(), output);
                return InputStatus.MORE_AVAILABLE;
            }
        }
        // 从 Checkpoint InputGate 读 CheckpointBarrier
        Optional<BufferOrEvent> bufferOrEvent = checkpointedInputGate.pollNext();
        ...
    }
}

@Override
public Optional<BufferOrEvent> pollNext() throws Exception {
    while (true) {

        ....
        Optional<BufferOrEvent> next = inputGate.pollNext();

        else if (bufferOrEvent.getEvent().getClass() == CheckpointBarrier.class) {
            CheckpointBarrier checkpointBarrier = (CheckpointBarrier) bufferOrEvent.getEvent();
            // CheckpointBarrier 的处理流程
            barrierHandler.processBarrier(checkpointBarrier, bufferOrEvent.getChannelInfo());
            return next;
        }
    ...
    }
}

/**
 * State for nextTransactionalIdHint.
 */
private transient ListState<FlinkKafkaProducer.NextTransactionalIdHint> nextTransactionalIdHintState;

@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
    super.snapshotState(context);

    nextTransactionalIdHintState.clear();
    // To avoid duplication only first subtask keeps track of next transactional id hint. Otherwise all of the
    // subtasks would write exactly same information.
    if (getRuntimeContext().getIndexOfThisSubtask() == 0 && semantic == FlinkKafkaProducer.Semantic.EXACTLY_ONCE) {
        checkState(nextTransactionalIdHint != null, "nextTransactionalIdHint must be set for EXACTLY_ONCE");
        long nextFreeTransactionalId = nextTransactionalIdHint.nextFreeTransactionalId;

        // If we scaled up, some (unknown) subtask must have created new transactional ids from scratch. In that
        // case we adjust nextFreeTransactionalId by the range of transactionalIds that could be used for this
        // scaling up.
        if (getRuntimeContext().getNumberOfParallelSubtasks() > nextTransactionalIdHint.lastParallelism) {
            nextFreeTransactionalId += getRuntimeContext().getNumberOfParallelSubtasks() * kafkaProducersPoolSize;
        }
        // 精确一次语义的 Checkpoint 状态
        nextTransactionalIdHintState.add(new FlinkKafkaProducer.NextTransactionalIdHint(
            getRuntimeContext().getNumberOfParallelSubtasks(),
            nextFreeTransactionalId));
    }
}

protected transient ListState<State<TXN, CONTEXT>> state;

@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
    // this is like the pre-commit of a 2-phase-commit transaction
    // we are ready to commit and remember the transaction

    checkState(currentTransactionHolder != null, "bug: no transaction object when performing state snapshot");

    long checkpointId = context.getCheckpointId();
    LOG.debug("{} - checkpoint {} triggered, flushing transaction '{}'", name(), context.getCheckpointId(), currentTransactionHolder);
    // 预提交即调用 producer.flush 提交数据到 Kafka
    preCommit(currentTransactionHolder.handle);
    pendingCommitTransactions.put(checkpointId, currentTransactionHolder);
    LOG.debug("{} - stored pending transactions {}", name(), pendingCommitTransactions);
    //     开启事务
    currentTransactionHolder = beginTransactionInternal();
    LOG.debug("{} - started new transaction '{}'", name(), currentTransactionHolder);

    state.clear();
    // 将 事务信息写入 状态
    state.add(new State<>(
        this.currentTransactionHolder,
        new ArrayList<>(pendingCommitTransactions.values()),
        userContext));
}

private void sendAcknowledgeMessages(long checkpointId, long timestamp) {
    // commit tasks
    for (ExecutionVertex ev : tasksToCommitTo) {
        Execution ee = ev.getCurrentExecutionAttempt();
        if (ee != null) {
            ee.notifyCheckpointComplete(checkpointId, timestamp);
        }
    }

    // commit coordinators
    for (OperatorCoordinatorCheckpointContext coordinatorContext : coordinatorsToCheckpoint) {
        coordinatorContext.checkpointComplete(checkpointId);
    }
}

@Override
protected void commit(FlinkKafkaProducer.KafkaTransactionState transaction) {
    if (transaction.isTransactional()) {
        try {
            // 调用 KafkaProduce 的方法，提交事务
            transaction.producer.commitTransaction();
        } finally {
            recycleTransactionalProducer(transaction.producer);
        }
    }
}