Flink Context到底是什么？

  • 2020 年 2 月 16 日
  • 筆記

Context ，又称执行上下文，特别抽象的一个东西，今天特地记录一下 Flink Context 到底是什么？有什么作用？不至于每天使用 Flink，总感觉云里雾里的

Flink Context 总共可以分为三种：StreamExecutionEnvironment、RuntimeContext、函数专有的Context

我们先看第一类：StreamExecutionEnvironment StreamExecutionEnvironment 包括 LocalStreamEnvironment、RemoteStreamEnvironment、StreamContextEnvironment。 我们在写 Flink 程序的时候，总会有

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

这一句话就是获得了 Flink 程序执行的上下文。具体的上下文又可以包括什么呢？

/** The default name to use for a streaming job if no other name has been specified. */  	public static final String DEFAULT_JOB_NAME = "Flink Streaming Job";    	/** The time characteristic that is used if none other is set. */  	private static final TimeCharacteristic DEFAULT_TIME_CHARACTERISTIC = TimeCharacteristic.ProcessingTime;    	/** The default buffer timeout (max delay of records in the network stack). */  	private static final long DEFAULT_NETWORK_BUFFER_TIMEOUT = 100L;    	/**  	 * The environment of the context (local by default, cluster if invoked through command line).  	 */  	private static StreamExecutionEnvironmentFactory contextEnvironmentFactory;    	/** The default parallelism used when creating a local environment. */  	private static int defaultLocalParallelism = Runtime.getRuntime().availableProcessors();    	// ------------------------------------------------------------------------    	/** The execution configuration for this environment. */  	private final ExecutionConfig config = new ExecutionConfig();    	/** Settings that control the checkpointing behavior. */  	private final CheckpointConfig checkpointCfg = new CheckpointConfig();    	protected final List<StreamTransformation<?>> transformations = new ArrayList<>();    	private long bufferTimeout = DEFAULT_NETWORK_BUFFER_TIMEOUT;    	protected boolean isChainingEnabled = true;    	/** The state backend used for storing k/v state and state snapshots. */  	private StateBackend defaultStateBackend;    	/** The time characteristic used by the data streams. */  	private TimeCharacteristic timeCharacteristic = DEFAULT_TIME_CHARACTERISTIC;

主要也就是包括 执行时配置 ExecutionConfig ，比如，我们熟悉的parallelism、maxParallelism等，还包括 CheckpointConfig 比如，checkpointTimeout、checkpointInterval等，还有 StateBackend、bufferTimeout( 后面会说 )，基本上包括了 Flink 程序执行所需的一切配置。

2. RuntimeContext 换记得吗？我们是怎么获取 state 的

listState = getRuntimeContext().getListState(kuduErrorDescriptor);

getRuntimeContext()得到的就是 RuntimeContext。 如果说 StreamExecutionEnvironment 是 Flink 程序之前必须的环境，那么 RuntimeContext 就是 Flink 程序执行中所必须的环境，每一个 RichFunction 都会有一个 RuntimeContext。 可以获得

String getTaskName();  int getIndexOfThisSubtask();  ExecutionConfig getExecutionConfig();  ClassLoader getUserCodeClassLoader();  IntCounter getIntCounter(String name);  <RT> List<RT> getBroadcastVariable(String name);  ...

**3.函数自己单独的 context 当我们定义一些 process Function 时，就经常会见到类似这样的函数

@Override  	public void processElement(Tuple2<String, Object> stringObjectTuple2, Context context, Collector<Tuple2<String, String>> collector) throws Exception {}

这个context究竟是什么呢？我们以 keyedProcessFunction 为例。

public abstract class Context {    		/**  		 * Timestamp of the element currently being processed or timestamp of a firing timer.  		 *  		 * <p>This might be {@code null}, for example if the time characteristic of your program  		 * is set to {@link org.apache.flink.streaming.api.TimeCharacteristic#ProcessingTime}.  		 */  		public abstract Long timestamp();    		/**  		 * A {@link TimerService} for querying time and registering timers.  		 */  		public abstract TimerService timerService();    		/**  		 还记得侧输出吗？  		 */  		public abstract <X> void output(OutputTag<X> outputTag, X value);    		/**  		当前处理的 key  		 */  		public abstract K getCurrentKey();  	}

可以得到 当前处理 element 的时间戳或者是 firing timer 的时间戳，还有 timerService，侧输出，当前正在处理的 key 等。

VirMach 便宜 VPS

QNews

QNews