@Public public interface RuntimeContext
A function can, during runtime, obtain the RuntimeContext via a call to AbstractRichFunction.getRuntimeContext()
.
Modifier and Type | Method and Description |
---|---|
<V,A extends Serializable> |
addAccumulator(String name,
Accumulator<V,A> accumulator)
Add this accumulator.
|
<V,A extends Serializable> |
getAccumulator(String name)
Get an existing accumulator object.
|
<IN,ACC,OUT> |
getAggregatingState(AggregatingStateDescriptor<IN,ACC,OUT> stateProperties)
Gets a handle to the system's key/value aggregating state.
|
int |
getAttemptNumber()
Gets the attempt number of this parallel subtask.
|
<RT> List<RT> |
getBroadcastVariable(String name)
Returns the result bound to the broadcast variable identified by the given
name . |
<T,C> C |
getBroadcastVariableWithInitializer(String name,
BroadcastVariableInitializer<T,C> initializer)
Returns the result bound to the broadcast variable identified by the given
name . |
DistributedCache |
getDistributedCache()
Returns the
DistributedCache to get the local temporary file copies of files
otherwise not locally accessible. |
DoubleCounter |
getDoubleCounter(String name)
Convenience function to create a counter object for doubles.
|
ExecutionConfig |
getExecutionConfig()
Returns the
ExecutionConfig for the currently executing
job. |
Set<ExternalResourceInfo> |
getExternalResourceInfos(String resourceName)
Get the specific external resource information by the resourceName.
|
Histogram |
getHistogram(String name)
Convenience function to create a counter object for histograms.
|
int |
getIndexOfThisSubtask()
Gets the number of this parallel subtask.
|
IntCounter |
getIntCounter(String name)
Convenience function to create a counter object for integers.
|
<T> ListState<T> |
getListState(ListStateDescriptor<T> stateProperties)
Gets a handle to the system's key/value list state.
|
LongCounter |
getLongCounter(String name)
Convenience function to create a counter object for longs.
|
<UK,UV> MapState<UK,UV> |
getMapState(MapStateDescriptor<UK,UV> stateProperties)
Gets a handle to the system's key/value map state.
|
int |
getMaxNumberOfParallelSubtasks()
Gets the number of max-parallelism with which the parallel task runs.
|
MetricGroup |
getMetricGroup()
Returns the metric group for this parallel subtask.
|
int |
getNumberOfParallelSubtasks()
Gets the parallelism with which the parallel task runs.
|
<T> ReducingState<T> |
getReducingState(ReducingStateDescriptor<T> stateProperties)
Gets a handle to the system's key/value reducing state.
|
<T> ValueState<T> |
getState(ValueStateDescriptor<T> stateProperties)
Gets a handle to the system's key/value state.
|
String |
getTaskName()
Returns the name of the task in which the UDF runs, as assigned during plan construction.
|
String |
getTaskNameWithSubtasks()
Returns the name of the task, appended with the subtask indicator, such as "MyTask (3/6)#1",
where 3 would be (
getIndexOfThisSubtask() + 1), and 6 would be getNumberOfParallelSubtasks() , and 1 would be getAttemptNumber() . |
ClassLoader |
getUserCodeClassLoader()
Gets the ClassLoader to load classes that are not in system's classpath, but are part of the
jar file of a user job.
|
boolean |
hasBroadcastVariable(String name)
Tests for the existence of the broadcast variable identified by the given
name . |
void |
registerUserCodeClassLoaderReleaseHookIfAbsent(String releaseHookName,
Runnable releaseHook)
Registers a custom hook for the user code class loader release.
|
String getTaskName()
@PublicEvolving MetricGroup getMetricGroup()
int getNumberOfParallelSubtasks()
@PublicEvolving int getMaxNumberOfParallelSubtasks()
int getIndexOfThisSubtask()
getNumberOfParallelSubtasks()
).int getAttemptNumber()
String getTaskNameWithSubtasks()
getIndexOfThisSubtask()
+ 1), and 6 would be getNumberOfParallelSubtasks()
, and 1 would be getAttemptNumber()
.ExecutionConfig getExecutionConfig()
ExecutionConfig
for the currently executing
job.ClassLoader getUserCodeClassLoader()
@PublicEvolving void registerUserCodeClassLoaderReleaseHookIfAbsent(String releaseHookName, Runnable releaseHook)
The release hook is executed just before the user code class loader is being released. Registration only happens if no hook has been registered under this name already.
releaseHookName
- name of the release hook.releaseHook
- release hook which is executed just before the user code class loader is
being released<V,A extends Serializable> void addAccumulator(String name, Accumulator<V,A> accumulator)
<V,A extends Serializable> Accumulator<V,A> getAccumulator(String name)
Throws an exception if the accumulator does not exist or if the accumulator exists, but with different type.
@PublicEvolving IntCounter getIntCounter(String name)
@PublicEvolving LongCounter getLongCounter(String name)
@PublicEvolving DoubleCounter getDoubleCounter(String name)
@PublicEvolving Histogram getHistogram(String name)
@PublicEvolving Set<ExternalResourceInfo> getExternalResourceInfos(String resourceName)
resourceName
- of the required external resource@PublicEvolving boolean hasBroadcastVariable(String name)
name
.name
- The name under which the broadcast variable is registered;<RT> List<RT> getBroadcastVariable(String name)
name
.
IMPORTANT: The broadcast variable data structure is shared between the parallel tasks on one machine. Any access that modifies its internal state needs to be manually synchronized by the caller.
name
- The name under which the broadcast variable is registered;<T,C> C getBroadcastVariableWithInitializer(String name, BroadcastVariableInitializer<T,C> initializer)
name
. The
broadcast variable is returned as a shared data structure that is initialized with the given
BroadcastVariableInitializer
.
IMPORTANT: The broadcast variable data structure is shared between the parallel tasks on one machine. Any access that modifies its internal state needs to be manually synchronized by the caller.
name
- The name under which the broadcast variable is registered;initializer
- The initializer that creates the shared data structure of the broadcast
variable from the sequence of elements.DistributedCache getDistributedCache()
DistributedCache
to get the local temporary file copies of files
otherwise not locally accessible.@PublicEvolving <T> ValueState<T> getState(ValueStateDescriptor<T> stateProperties)
Because the scope of each value is the key of the currently processed element, and the elements are distributed by the Flink runtime, the system can transparently scale out and redistribute the state and KeyedStream.
The following code example shows how to implement a continuous counter that counts how many times elements of a certain key occur, and emits an updated count for that element on each occurrence.
DataStream<MyType> stream = ...;
KeyedStream<MyType> keyedStream = stream.keyBy("id");
keyedStream.map(new RichMapFunction<MyType, Tuple2<MyType, Long>>() {
private ValueState<Long> state;
public void open(Configuration cfg) {
state = getRuntimeContext().getState(
new ValueStateDescriptor<Long>("count", LongSerializer.INSTANCE, 0L));
}
public Tuple2<MyType, Long> map(MyType value) {
long count = state.value() + 1;
state.update(count);
return new Tuple2<>(value, count);
}
});
T
- The type of value stored in the state.stateProperties
- The descriptor defining the properties of the stats.UnsupportedOperationException
- Thrown, if no partitioned state is available for the
function (function is not part of a KeyedStream).@PublicEvolving <T> ListState<T> getListState(ListStateDescriptor<T> stateProperties)
getState(ValueStateDescriptor)
, but is optimized for state that holds
lists. One can add elements to the list, or retrieve the list as a whole.
This state is only accessible if the function is executed on a KeyedStream.
DataStream<MyType> stream = ...;
KeyedStream<MyType> keyedStream = stream.keyBy("id");
keyedStream.map(new RichFlatMapFunction<MyType, List<MyType>>() {
private ListState<MyType> state;
public void open(Configuration cfg) {
state = getRuntimeContext().getListState(
new ListStateDescriptor<>("myState", MyType.class));
}
public void flatMap(MyType value, Collector<MyType> out) {
if (value.isDivider()) {
for (MyType t : state.get()) {
out.collect(t);
}
} else {
state.add(value);
}
}
});
T
- The type of value stored in the state.stateProperties
- The descriptor defining the properties of the stats.UnsupportedOperationException
- Thrown, if no partitioned state is available for the
function (function is not part os a KeyedStream).@PublicEvolving <T> ReducingState<T> getReducingState(ReducingStateDescriptor<T> stateProperties)
getState(ValueStateDescriptor)
, but is optimized for state that
aggregates values.
This state is only accessible if the function is executed on a KeyedStream.
DataStream<MyType> stream = ...;
KeyedStream<MyType> keyedStream = stream.keyBy("id");
keyedStream.map(new RichMapFunction<MyType, List<MyType>>() {
private ReducingState<Long> state;
public void open(Configuration cfg) {
state = getRuntimeContext().getReducingState(
new ReducingStateDescriptor<>("sum", (a, b) -> a + b, Long.class));
}
public Tuple2<MyType, Long> map(MyType value) {
state.add(value.count());
return new Tuple2<>(value, state.get());
}
});
T
- The type of value stored in the state.stateProperties
- The descriptor defining the properties of the stats.UnsupportedOperationException
- Thrown, if no partitioned state is available for the
function (function is not part of a KeyedStream).@PublicEvolving <IN,ACC,OUT> AggregatingState<IN,OUT> getAggregatingState(AggregatingStateDescriptor<IN,ACC,OUT> stateProperties)
getState(ValueStateDescriptor)
, but is optimized for state that
aggregates values with different types.
This state is only accessible if the function is executed on a KeyedStream.
DataStream<MyType> stream = ...;
KeyedStream<MyType> keyedStream = stream.keyBy("id");
AggregateFunction<...> aggregateFunction = ...
keyedStream.map(new RichMapFunction<MyType, List<MyType>>() {
private AggregatingState<MyType, Long> state;
public void open(Configuration cfg) {
state = getRuntimeContext().getAggregatingState(
new AggregatingStateDescriptor<>("sum", aggregateFunction, Long.class));
}
public Tuple2<MyType, Long> map(MyType value) {
state.add(value);
return new Tuple2<>(value, state.get());
}
});
IN
- The type of the values that are added to the state.ACC
- The type of the accumulator (intermediate aggregation state).OUT
- The type of the values that are returned from the state.stateProperties
- The descriptor defining the properties of the stats.UnsupportedOperationException
- Thrown, if no partitioned state is available for the
function (function is not part of a KeyedStream).@PublicEvolving <UK,UV> MapState<UK,UV> getMapState(MapStateDescriptor<UK,UV> stateProperties)
getState(ValueStateDescriptor)
, but is optimized for state that is
composed of user-defined key-value pairs
This state is only accessible if the function is executed on a KeyedStream.
DataStream<MyType> stream = ...;
KeyedStream<MyType> keyedStream = stream.keyBy("id");
keyedStream.map(new RichMapFunction<MyType, List<MyType>>() {
private MapState<MyType, Long> state;
public void open(Configuration cfg) {
state = getRuntimeContext().getMapState(
new MapStateDescriptor<>("sum", MyType.class, Long.class));
}
public Tuple2<MyType, Long> map(MyType value) {
return new Tuple2<>(value, state.get(value));
}
});
UK
- The type of the user keys stored in the state.UV
- The type of the user values stored in the state.stateProperties
- The descriptor defining the properties of the stats.UnsupportedOperationException
- Thrown, if no partitioned state is available for the
function (function is not part of a KeyedStream).Copyright © 2014–2021 The Apache Software Foundation. All rights reserved.