Flink 源码剖析｜RuntimeContext 接口-Toy模板网

这篇具有很好参考价值的文章主要介绍了Flink 源码剖析｜RuntimeContext 接口。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

每个并行的实例都会包含一个 RuntimeContext。RuntimeContext 接口包含函数执行的上下文信息，提供了如下功能：

访问静态上下文信息（例如当前并行度）
添加及访问累加器
访问外部资源信息
访问广播变量和分布式缓存
访问并编辑状态

下面，我们逐类介绍 RuntimeContext 接口的方法。

Flink 源码剖析｜RuntimeContext 接口,Flink,flink,RuntimeContext

1 访问静态上下文信息

RuntimeContext 的包括如下获取静态上下文信息的方法：

getJobId()：获取当前作业的 ID。
getTaskName()：获取执行 UDF 的 task 名称。
getMetricGroup()：获取当前 subtask 的指标组。
getNumberOfParallelSubtasks()：获取执行 UDF 的 task 的并行度。
getMaxNumberOfParallelSubtasks()：获取执行 UDF 的 task 的最大并行度。
getIndexOfThisSubtask()：获取执行 UDF 的 subtask 的编号（编号从 0 开始）。
getAttemptNumber()：获取执行 UDF 的 subtask 的尝试次数（第一次尝试的次数是 0）。
getTaskNameWithSubtasks()：获取执行 UDF 的 subtask 的名称。这个名称如下 {任务名称} ({subtask的编号}/{并行度})#{尝试次数}，例如 MyTask (3/6)#1，其中 {任务名称} 为 getTaskName() 的返回值，{subtask的编号} 为 getIndexOfThisSubtask() + 1，{并行度} 为 getNumberOfParallelSubtasks() 的返回值，{尝试次数} 为 getAttemptNumber() 的返回值。
getExecutionConfig()：获取当前作业的执行配置。
createSerializer(TypeInformation<T> typeInformation)：获取指定类型的序列化器。
getGlobalJobParameters()：获取作业的全局参数。
isObjectReuseEnabled()：获取对象重用是否开启。
getUserCodeClassLoader()：获取用户类（不在系统 classpath 的类）的加载器。
registerUserCodeClassLoaderReleaseHookIfAbsent(String releaseHookName, Runnable releaseHook)：在用户类加载器中注册自定义 hook。

大概可以归纳为如下几种类型

作业、task、subtask 的基本信息及尝试次数
指标组信息
包括序列化器和类加载器在内的配置信息

源码｜Github｜org.apache.flink.api.common.functions.RuntimeContext（部分）文章来源地址https://www.toymoban.com/news/detail-835339.html

JobID getJobId();

String getTaskName();

@PublicEvolving
OperatorMetricGroup getMetricGroup();

int getNumberOfParallelSubtasks();

@PublicEvolving
int getMaxNumberOfParallelSubtasks();

int getIndexOfThisSubtask();

int getAttemptNumber();

String getTaskNameWithSubtasks();

@Deprecated
ExecutionConfig getExecutionConfig();

@PublicEvolving
<T> TypeSerializer<T> createSerializer(TypeInformation<T> typeInformation);

@PublicEvolving
Map<String, String> getGlobalJobParameters();

@PublicEvolving
boolean isObjectReuseEnabled();

ClassLoader getUserCodeClassLoader();

@PublicEvolving
void registerUserCodeClassLoaderReleaseHookIfAbsent(
        String releaseHookName, Runnable releaseHook);

2 添加及访问累加器

RuntimeContext 的包括如下累加器相关的方法：

addAccumulator(String name, Accumulator<V, A> accumulator)：添加累加器
Accumulator<V, A> getAccumulator(String name)：获取基础类型的累加器
getIntCounter(String name)：获取 IntCounter 类型的累加器
getLongCounter(String name)：获取 LongCounter 类型的累加器
getDoubleCounter(String name)：获取 DoubleCounter 类型的累加器
getHistogram(String name)：获取直方图类型的累加器

源码｜Github｜org.apache.flink.api.common.functions.RuntimeContext（部分）

<V, A extends Serializable> void addAccumulator(String name, Accumulator<V, A> accumulator);

<V, A extends Serializable> Accumulator<V, A> getAccumulator(String name);

@PublicEvolving
IntCounter getIntCounter(String name);

@PublicEvolving
LongCounter getLongCounter(String name);

@PublicEvolving
DoubleCounter getDoubleCounter(String name);

@PublicEvolving
Histogram getHistogram(String name);

3 访问外部资源

RuntimeContext 提供了 getExternalResourceInfos(String resourceName) 外部资源信息。

源码｜Github｜org.apache.flink.api.common.functions.RuntimeContext#getExternalResourceInfos

@PublicEvolving
Set<ExternalResourceInfo> getExternalResourceInfos(String resourceName);

4 访问广播变量

RuntimeContext 提供了如下访问广播变量的方法：

hasBroadcastVariable(String name)：检查是否包含名称为 name 的广播变量
getBroadcastVariable(String name)：返回名称为 name 的广播变量
getBroadcastVariableWithInitializer(String name, BroadcastVariableInitializer<T, C> initializer)：返回名称为 name 的广播变量，并使用 initializer 初始化。
getDistributedCache()：访问分布式缓存。

源码｜Github｜org.apache.flink.api.common.functions.RuntimeContext（部分）

@PublicEvolving
boolean hasBroadcastVariable(String name);

<RT> List<RT> getBroadcastVariable(String name);

<T, C> C getBroadcastVariableWithInitializer(
        String name, BroadcastVariableInitializer<T, C> initializer);

DistributedCache getDistributedCache();

5 访问并编辑状态

RuntimeContext 提供了如下方法用于获取不同类型的状态对象，在获取到状态对象后，可以对状态对象进行编辑：

getState(ValueStateDescriptor<T> stateProperties)：获取 ValueState 类型的状态对象
getListState(ListStateDescriptor<T> stateProperties)：获取 ListState 类型的状态对象
getReducingState(ReducingStateDescriptor<T> stateProperties)：获取 ReducingState 类型的状态对象
getAggregatingState(AggregatingStateDescriptor<IN, ACC, OUT> stateProperties)：获取 AggregatingState 类型的状态对象
getMapState(MapStateDescriptor<UK, UV> stateProperties)：获取 MapState 类型的状态对象

源码｜Github｜org.apache.flink.api.common.functions.RuntimeContext（部分）

@PublicEvolving
<T> ValueState<T> getState(ValueStateDescriptor<T> stateProperties);

@PublicEvolving
<T> ListState<T> getListState(ListStateDescriptor<T> stateProperties);

@PublicEvolving
<T> ReducingState<T> getReducingState(ReducingStateDescriptor<T> stateProperties);

@PublicEvolving
<IN, ACC, OUT> AggregatingState<IN, OUT> getAggregatingState(
        AggregatingStateDescriptor<IN, ACC, OUT> stateProperties);

@PublicEvolving
<UK, UV> MapState<UK, UV> getMapState(MapStateDescriptor<UK, UV> stateProperties);