| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.flink.runtime.operators; |
| |
| import org.apache.flink.api.common.ExecutionConfig; |
| import org.apache.flink.api.common.accumulators.Accumulator; |
| import org.apache.flink.api.common.distributions.DataDistribution; |
| import org.apache.flink.api.common.functions.DefaultOpenContext; |
| import org.apache.flink.api.common.functions.Function; |
| import org.apache.flink.api.common.functions.GroupCombineFunction; |
| import org.apache.flink.api.common.functions.OpenContext; |
| import org.apache.flink.api.common.functions.Partitioner; |
| import org.apache.flink.api.common.functions.util.FunctionUtils; |
| import org.apache.flink.api.common.typeutils.TypeComparator; |
| import org.apache.flink.api.common.typeutils.TypeComparatorFactory; |
| import org.apache.flink.api.common.typeutils.TypeSerializerFactory; |
| import org.apache.flink.configuration.Configuration; |
| import org.apache.flink.core.memory.MemorySegment; |
| import org.apache.flink.metrics.groups.OperatorMetricGroup; |
| import org.apache.flink.runtime.broadcast.BroadcastVariableMaterialization; |
| import org.apache.flink.runtime.execution.CancelTaskException; |
| import org.apache.flink.runtime.execution.Environment; |
| import org.apache.flink.runtime.io.disk.iomanager.IOManager; |
| import org.apache.flink.runtime.io.network.api.reader.MutableReader; |
| import org.apache.flink.runtime.io.network.api.reader.MutableRecordReader; |
| import org.apache.flink.runtime.io.network.api.writer.ChannelSelector; |
| import org.apache.flink.runtime.io.network.api.writer.RecordWriter; |
| import org.apache.flink.runtime.io.network.api.writer.RecordWriterBuilder; |
| import org.apache.flink.runtime.io.network.partition.consumer.IndexedInputGate; |
| import org.apache.flink.runtime.io.network.partition.consumer.UnionInputGate; |
| import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable; |
| import org.apache.flink.runtime.memory.MemoryManager; |
| import org.apache.flink.runtime.metrics.groups.InternalOperatorMetricGroup; |
| import org.apache.flink.runtime.operators.chaining.ChainedDriver; |
| import org.apache.flink.runtime.operators.chaining.ExceptionInChainedStubException; |
| import org.apache.flink.runtime.operators.resettable.SpillingResettableMutableObjectIterator; |
| import org.apache.flink.runtime.operators.shipping.OutputCollector; |
| import org.apache.flink.runtime.operators.shipping.OutputEmitter; |
| import org.apache.flink.runtime.operators.shipping.ShipStrategyType; |
| import org.apache.flink.runtime.operators.sort.ExternalSorter; |
| import org.apache.flink.runtime.operators.sort.Sorter; |
| import org.apache.flink.runtime.operators.util.CloseableInputProvider; |
| import org.apache.flink.runtime.operators.util.DistributedRuntimeUDFContext; |
| import org.apache.flink.runtime.operators.util.LocalStrategy; |
| import org.apache.flink.runtime.operators.util.ReaderIterator; |
| import org.apache.flink.runtime.operators.util.TaskConfig; |
| import org.apache.flink.runtime.plugable.DeserializationDelegate; |
| import org.apache.flink.runtime.plugable.SerializationDelegate; |
| import org.apache.flink.runtime.taskmanager.TaskManagerRuntimeInfo; |
| import org.apache.flink.util.Collector; |
| import org.apache.flink.util.InstantiationUtil; |
| import org.apache.flink.util.MutableObjectIterator; |
| import org.apache.flink.util.Preconditions; |
| import org.apache.flink.util.UserCodeClassLoader; |
| |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.List; |
| import java.util.Map; |
| |
| import static java.util.Collections.emptyList; |
| |
| /** |
| * The base class for all batch tasks. Encapsulated common behavior and implements the main |
| * life-cycle of the user code. |
| */ |
| public class BatchTask<S extends Function, OT> extends AbstractInvokable |
| implements TaskContext<S, OT> { |
| |
| protected static final Logger LOG = LoggerFactory.getLogger(BatchTask.class); |
| |
| // -------------------------------------------------------------------------------------------- |
| |
| /** |
| * The driver that invokes the user code (the stub implementation). The central driver in this |
| * task (further drivers may be chained behind this driver). |
| */ |
| protected volatile Driver<S, OT> driver; |
| |
| /** |
| * The instantiated user code of this task's main operator (driver). May be null if the operator |
| * has no udf. |
| */ |
| protected S stub; |
| |
| /** The udf's runtime context. */ |
| protected DistributedRuntimeUDFContext runtimeUdfContext; |
| |
| /** |
| * The collector that forwards the user code's results. May forward to a channel or to chained |
| * drivers within this task. |
| */ |
| protected Collector<OT> output; |
| |
| /** |
| * The output writers for the data that this task forwards to the next task. The latest driver |
| * (the central, if no chained drivers exist, otherwise the last chained driver) produces its |
| * output to these writers. |
| */ |
| protected List<RecordWriter<?>> eventualOutputs; |
| |
| /** The input readers of this task. */ |
| protected MutableReader<?>[] inputReaders; |
| |
| /** The input readers for the configured broadcast variables for this task. */ |
| protected MutableReader<?>[] broadcastInputReaders; |
| |
| /** The inputs reader, wrapped in an iterator. Prior to the local strategies, etc... */ |
| protected MutableObjectIterator<?>[] inputIterators; |
| |
| /** The indices of the iterative inputs. Empty, if the task is not iterative. */ |
| protected int[] iterativeInputs; |
| |
| /** The indices of the iterative broadcast inputs. Empty, if non of the inputs is iterative. */ |
| protected int[] iterativeBroadcastInputs; |
| |
| /** The local strategies that are applied on the inputs. */ |
| protected volatile CloseableInputProvider<?>[] localStrategies; |
| |
| /** |
| * The optional temp barriers on the inputs for dead-lock breaking. Are optionally resettable. |
| */ |
| protected volatile TempBarrier<?>[] tempBarriers; |
| |
| /** The resettable inputs in the case where no temp barrier is needed. */ |
| protected volatile SpillingResettableMutableObjectIterator<?>[] resettableInputs; |
| |
| /** |
| * The inputs to the operator. Return the readers' data after the application of the local |
| * strategy and the temp-table barrier. |
| */ |
| protected MutableObjectIterator<?>[] inputs; |
| |
| /** The serializers for the input data type. */ |
| protected TypeSerializerFactory<?>[] inputSerializers; |
| |
| /** The serializers for the broadcast input data types. */ |
| protected TypeSerializerFactory<?>[] broadcastInputSerializers; |
| |
| /** The comparators for the central driver. */ |
| protected TypeComparator<?>[] inputComparators; |
| |
| /** The task configuration with the setup parameters. */ |
| protected TaskConfig config; |
| |
| /** A list of chained drivers, if there are any. */ |
| protected ArrayList<ChainedDriver<?, ?>> chainedTasks; |
| |
| /** |
| * Certain inputs may be excluded from resetting. For example, the initial partial solution in |
| * an iteration head must not be reset (it is read through the back channel), when all others |
| * are reset. |
| */ |
| private boolean[] excludeFromReset; |
| |
| /** Flag indicating for each input whether it is cached and can be reset. */ |
| private boolean[] inputIsCached; |
| |
| /** flag indicating for each input whether it must be asynchronously materialized. */ |
| private boolean[] inputIsAsyncMaterialized; |
| |
| /** The amount of memory per input that is dedicated to the materialization. */ |
| private int[] materializationMemory; |
| |
| /** The flag that tags the task as still running. Checked periodically to abort processing. */ |
| protected volatile boolean running = true; |
| |
| /** The accumulator map used in the RuntimeContext. */ |
| protected Map<String, Accumulator<?, ?>> accumulatorMap; |
| |
| private InternalOperatorMetricGroup metrics; |
| |
| // -------------------------------------------------------------------------------------------- |
| // Constructor |
| // -------------------------------------------------------------------------------------------- |
| |
| /** |
| * Create an Invokable task and set its environment. |
| * |
| * @param environment The environment assigned to this invokable. |
| */ |
| public BatchTask(Environment environment) { |
| super(environment); |
| } |
| |
| // -------------------------------------------------------------------------------------------- |
| // Task Interface |
| // -------------------------------------------------------------------------------------------- |
| |
| /** The main work method. */ |
| @Override |
| public void invoke() throws Exception { |
| // -------------------------------------------------------------------- |
| // Initialize |
| // -------------------------------------------------------------------- |
| if (LOG.isDebugEnabled()) { |
| LOG.debug(formatLogString("Start registering input and output.")); |
| } |
| |
| // obtain task configuration (including stub parameters) |
| Configuration taskConf = getTaskConfiguration(); |
| this.config = new TaskConfig(taskConf); |
| |
| // now get the operator class which drives the operation |
| final Class<? extends Driver<S, OT>> driverClass = this.config.getDriver(); |
| this.driver = InstantiationUtil.instantiate(driverClass, Driver.class); |
| |
| String headName = getEnvironment().getTaskInfo().getTaskName().split("->")[0].trim(); |
| this.metrics = |
| getEnvironment() |
| .getMetricGroup() |
| .getOrAddOperator( |
| headName.startsWith("CHAIN") ? headName.substring(6) : headName); |
| this.metrics.getIOMetricGroup().reuseInputMetricsForTask(); |
| if (config.getNumberOfChainedStubs() == 0) { |
| this.metrics.getIOMetricGroup().reuseOutputMetricsForTask(); |
| } |
| |
| // initialize the readers. |
| // this does not yet trigger any stream consuming or processing. |
| initInputReaders(); |
| initBroadcastInputReaders(); |
| |
| // initialize the writers. |
| initOutputs(); |
| |
| if (LOG.isDebugEnabled()) { |
| LOG.debug(formatLogString("Finished registering input and output.")); |
| } |
| |
| // -------------------------------------------------------------------- |
| // Invoke |
| // -------------------------------------------------------------------- |
| if (LOG.isDebugEnabled()) { |
| LOG.debug(formatLogString("Start task code.")); |
| } |
| |
| this.runtimeUdfContext = createRuntimeContext(metrics); |
| |
| // whatever happens in this scope, make sure that the local strategies are cleaned up! |
| // note that the initialization of the local strategies is in the try-finally block as well, |
| // so that the thread that creates them catches its own errors that may happen in that |
| // process. |
| // this is especially important, since there may be asynchronous closes (such as through |
| // canceling). |
| try { |
| // initialize the remaining data structures on the input and trigger the local |
| // processing |
| // the local processing includes building the dams / caches |
| try { |
| int numInputs = driver.getNumberOfInputs(); |
| int numComparators = driver.getNumberOfDriverComparators(); |
| int numBroadcastInputs = this.config.getNumBroadcastInputs(); |
| |
| initInputsSerializersAndComparators(numInputs, numComparators); |
| initBroadcastInputsSerializers(numBroadcastInputs); |
| |
| // set the iterative status for inputs and broadcast inputs |
| { |
| List<Integer> iterativeInputs = new ArrayList<>(); |
| |
| for (int i = 0; i < numInputs; i++) { |
| final int numberOfEventsUntilInterrupt = |
| getTaskConfig().getNumberOfEventsUntilInterruptInIterativeGate(i); |
| |
| if (numberOfEventsUntilInterrupt < 0) { |
| throw new IllegalArgumentException(); |
| } else if (numberOfEventsUntilInterrupt > 0) { |
| this.inputReaders[i].setIterativeReader(); |
| iterativeInputs.add(i); |
| |
| if (LOG.isDebugEnabled()) { |
| LOG.debug( |
| formatLogString( |
| "Input [" |
| + i |
| + "] reads in supersteps with [" |
| + numberOfEventsUntilInterrupt |
| + "] event(s) till next superstep.")); |
| } |
| } |
| } |
| this.iterativeInputs = asArray(iterativeInputs); |
| } |
| |
| { |
| List<Integer> iterativeBcInputs = new ArrayList<>(); |
| |
| for (int i = 0; i < numBroadcastInputs; i++) { |
| final int numberOfEventsUntilInterrupt = |
| getTaskConfig() |
| .getNumberOfEventsUntilInterruptInIterativeBroadcastGate(i); |
| |
| if (numberOfEventsUntilInterrupt < 0) { |
| throw new IllegalArgumentException(); |
| } else if (numberOfEventsUntilInterrupt > 0) { |
| this.broadcastInputReaders[i].setIterativeReader(); |
| iterativeBcInputs.add(i); |
| |
| if (LOG.isDebugEnabled()) { |
| LOG.debug( |
| formatLogString( |
| "Broadcast input [" |
| + i |
| + "] reads in supersteps with [" |
| + numberOfEventsUntilInterrupt |
| + "] event(s) till next superstep.")); |
| } |
| } |
| } |
| this.iterativeBroadcastInputs = asArray(iterativeBcInputs); |
| } |
| |
| initLocalStrategies(numInputs); |
| } catch (Exception e) { |
| throw new RuntimeException( |
| "Initializing the input processing failed" |
| + (e.getMessage() == null ? "." : ": " + e.getMessage()), |
| e); |
| } |
| |
| if (!this.running) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug(formatLogString("Task cancelled before task code was started.")); |
| } |
| return; |
| } |
| |
| // pre main-function initialization |
| initialize(); |
| |
| // read the broadcast variables. they will be released in the finally clause |
| for (int i = 0; i < this.config.getNumBroadcastInputs(); i++) { |
| final String name = this.config.getBroadcastInputName(i); |
| readAndSetBroadcastInput( |
| i, name, this.runtimeUdfContext, 1 /* superstep one for the start */); |
| } |
| |
| // the work goes here |
| run(); |
| } finally { |
| // clean up in any case! |
| closeLocalStrategiesAndCaches(); |
| |
| clearReaders(inputReaders); |
| clearWriters(eventualOutputs); |
| } |
| |
| if (this.running) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug(formatLogString("Finished task code.")); |
| } |
| } else { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug(formatLogString("Task code cancelled.")); |
| } |
| } |
| } |
| |
| @Override |
| public void cancel() throws Exception { |
| this.running = false; |
| |
| if (LOG.isDebugEnabled()) { |
| LOG.debug(formatLogString("Cancelling task code")); |
| } |
| |
| try { |
| if (this.driver != null) { |
| this.driver.cancel(); |
| } |
| } finally { |
| closeLocalStrategiesAndCaches(); |
| } |
| } |
| |
| // -------------------------------------------------------------------------------------------- |
| // Main Work Methods |
| // -------------------------------------------------------------------------------------------- |
| |
| protected void initialize() throws Exception { |
| // create the operator |
| try { |
| this.driver.setup(this); |
| } catch (Throwable t) { |
| throw new Exception( |
| "The driver setup for '" |
| + this.getEnvironment().getTaskInfo().getTaskName() |
| + "' , caused an error: " |
| + t.getMessage(), |
| t); |
| } |
| |
| // instantiate the UDF |
| try { |
| final Class<? super S> userCodeFunctionType = this.driver.getStubType(); |
| // if the class is null, the driver has no user code |
| if (userCodeFunctionType != null) { |
| this.stub = initStub(userCodeFunctionType); |
| } |
| } catch (Exception e) { |
| throw new RuntimeException( |
| "Initializing the UDF" + (e.getMessage() == null ? "." : ": " + e.getMessage()), |
| e); |
| } |
| } |
| |
| protected <X> void readAndSetBroadcastInput( |
| int inputNum, String bcVarName, DistributedRuntimeUDFContext context, int superstep) |
| throws IOException { |
| |
| if (LOG.isDebugEnabled()) { |
| LOG.debug( |
| formatLogString( |
| "Setting broadcast variable '" |
| + bcVarName |
| + "'" |
| + (superstep > 1 ? ", superstep " + superstep : ""))); |
| } |
| |
| @SuppressWarnings("unchecked") |
| final TypeSerializerFactory<X> serializerFactory = |
| (TypeSerializerFactory<X>) this.broadcastInputSerializers[inputNum]; |
| |
| final MutableReader<?> reader = this.broadcastInputReaders[inputNum]; |
| |
| BroadcastVariableMaterialization<X, ?> variable = |
| getEnvironment() |
| .getBroadcastVariableManager() |
| .materializeBroadcastVariable( |
| bcVarName, superstep, this, reader, serializerFactory); |
| context.setBroadcastVariable(bcVarName, variable); |
| } |
| |
| protected void releaseBroadcastVariables( |
| String bcVarName, int superstep, DistributedRuntimeUDFContext context) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug( |
| formatLogString( |
| "Releasing broadcast variable '" |
| + bcVarName |
| + "'" |
| + (superstep > 1 ? ", superstep " + superstep : ""))); |
| } |
| |
| getEnvironment().getBroadcastVariableManager().releaseReference(bcVarName, superstep, this); |
| context.clearBroadcastVariable(bcVarName); |
| } |
| |
| protected void run() throws Exception { |
| // ---------------------------- Now, the actual processing starts ------------------------ |
| // check for asynchronous canceling |
| if (!this.running) { |
| return; |
| } |
| |
| boolean stubOpen = false; |
| |
| try { |
| // run the data preparation |
| try { |
| this.driver.prepare(); |
| } catch (Throwable t) { |
| // if the preparation caused an error, clean up |
| // errors during clean-up are swallowed, because we have already a root exception |
| throw new Exception( |
| "The data preparation for task '" |
| + this.getEnvironment().getTaskInfo().getTaskName() |
| + "' , caused an error: " |
| + t.getMessage(), |
| t); |
| } |
| |
| // check for canceling |
| if (!this.running) { |
| return; |
| } |
| |
| // start all chained tasks |
| BatchTask.openChainedTasks(this.chainedTasks, this); |
| |
| // open stub implementation |
| if (this.stub != null) { |
| try { |
| Configuration stubConfig = this.config.getStubParameters(); |
| FunctionUtils.openFunction(this.stub, DefaultOpenContext.INSTANCE); |
| stubOpen = true; |
| } catch (Throwable t) { |
| throw new Exception( |
| "The user defined 'open()' method caused an exception: " |
| + t.getMessage(), |
| t); |
| } |
| } |
| |
| // run the user code |
| this.driver.run(); |
| |
| // close. We close here such that a regular close throwing an exception marks a task as |
| // failed. |
| if (this.running && this.stub != null) { |
| FunctionUtils.closeFunction(this.stub); |
| stubOpen = false; |
| } |
| |
| // close all chained tasks letting them report failure |
| BatchTask.closeChainedTasks(this.chainedTasks, this); |
| |
| // close the output collector |
| this.output.close(); |
| } catch (Exception ex) { |
| // close the input, but do not report any exceptions, since we already have another root |
| // cause |
| if (stubOpen) { |
| try { |
| FunctionUtils.closeFunction(this.stub); |
| } catch (Throwable t) { |
| // do nothing |
| } |
| } |
| |
| // if resettable driver invoke teardown |
| if (this.driver instanceof ResettableDriver) { |
| final ResettableDriver<?, ?> resDriver = (ResettableDriver<?, ?>) this.driver; |
| try { |
| resDriver.teardown(); |
| } catch (Throwable t) { |
| throw new Exception( |
| "Error while shutting down an iterative operator: " + t.getMessage(), |
| t); |
| } |
| } |
| |
| BatchTask.cancelChainedTasks(this.chainedTasks); |
| |
| ex = ExceptionInChainedStubException.exceptionUnwrap(ex); |
| |
| if (ex instanceof CancelTaskException) { |
| // forward canceling exception |
| throw ex; |
| } else if (this.running) { |
| // throw only if task was not cancelled. in the case of canceling, exceptions are |
| // expected |
| BatchTask.logAndThrowException(ex, this); |
| } |
| } finally { |
| this.driver.cleanup(); |
| } |
| } |
| |
| protected void closeLocalStrategiesAndCaches() { |
| |
| // make sure that all broadcast variable references held by this task are released |
| if (LOG.isDebugEnabled()) { |
| LOG.debug(formatLogString("Releasing all broadcast variables.")); |
| } |
| |
| getEnvironment().getBroadcastVariableManager().releaseAllReferencesFromTask(this); |
| if (runtimeUdfContext != null) { |
| runtimeUdfContext.clearAllBroadcastVariables(); |
| } |
| |
| // clean all local strategies and caches/pipeline breakers. |
| |
| if (this.localStrategies != null) { |
| for (int i = 0; i < this.localStrategies.length; i++) { |
| if (this.localStrategies[i] != null) { |
| try { |
| this.localStrategies[i].close(); |
| } catch (Throwable t) { |
| LOG.error("Error closing local strategy for input " + i, t); |
| } |
| } |
| } |
| } |
| if (this.tempBarriers != null) { |
| for (int i = 0; i < this.tempBarriers.length; i++) { |
| if (this.tempBarriers[i] != null) { |
| try { |
| this.tempBarriers[i].close(); |
| } catch (Throwable t) { |
| LOG.error("Error closing temp barrier for input " + i, t); |
| } |
| } |
| } |
| } |
| if (this.resettableInputs != null) { |
| for (int i = 0; i < this.resettableInputs.length; i++) { |
| if (this.resettableInputs[i] != null) { |
| try { |
| this.resettableInputs[i].close(); |
| } catch (Throwable t) { |
| LOG.error("Error closing cache for input " + i, t); |
| } |
| } |
| } |
| } |
| } |
| |
| // -------------------------------------------------------------------------------------------- |
| // Task Setup and Teardown |
| // -------------------------------------------------------------------------------------------- |
| |
| /** |
| * @return the last output collector in the collector chain |
| */ |
| @SuppressWarnings("unchecked") |
| protected Collector<OT> getLastOutputCollector() { |
| int numChained = this.chainedTasks.size(); |
| return (numChained == 0) |
| ? output |
| : (Collector<OT>) chainedTasks.get(numChained - 1).getOutputCollector(); |
| } |
| |
| /** |
| * Sets the last output {@link Collector} of the collector chain of this {@link BatchTask}. |
| * |
| * <p>In case of chained tasks, the output collector of the last {@link ChainedDriver} is set. |
| * Otherwise it is the single collector of the {@link BatchTask}. |
| * |
| * @param newOutputCollector new output collector to set as last collector |
| */ |
| protected void setLastOutputCollector(Collector<OT> newOutputCollector) { |
| int numChained = this.chainedTasks.size(); |
| |
| if (numChained == 0) { |
| output = newOutputCollector; |
| return; |
| } |
| |
| chainedTasks.get(numChained - 1).setOutputCollector(newOutputCollector); |
| } |
| |
| public TaskConfig getLastTasksConfig() { |
| int numChained = this.chainedTasks.size(); |
| return (numChained == 0) ? config : chainedTasks.get(numChained - 1).getTaskConfig(); |
| } |
| |
| protected S initStub(Class<? super S> stubSuperClass) throws Exception { |
| try { |
| ClassLoader userCodeClassLoader = getUserCodeClassLoader(); |
| S stub = |
| config.<S>getStubWrapper(userCodeClassLoader) |
| .getUserCodeObject(stubSuperClass, userCodeClassLoader); |
| // check if the class is a subclass, if the check is required |
| if (stubSuperClass != null && !stubSuperClass.isAssignableFrom(stub.getClass())) { |
| throw new RuntimeException( |
| "The class '" |
| + stub.getClass().getName() |
| + "' is not a subclass of '" |
| + stubSuperClass.getName() |
| + "' as is required."); |
| } |
| FunctionUtils.setFunctionRuntimeContext(stub, this.runtimeUdfContext); |
| return stub; |
| } catch (ClassCastException ccex) { |
| throw new Exception( |
| "The stub class is not a proper subclass of " + stubSuperClass.getName(), ccex); |
| } |
| } |
| |
| /** |
| * Creates the record readers for the number of inputs as defined by {@link |
| * #getNumTaskInputs()}. This method requires that the task configuration, the driver, and the |
| * user-code class loader are set. |
| */ |
| protected void initInputReaders() throws Exception { |
| final int numInputs = getNumTaskInputs(); |
| final MutableReader<?>[] inputReaders = new MutableReader<?>[numInputs]; |
| |
| int currentReaderOffset = 0; |
| |
| for (int i = 0; i < numInputs; i++) { |
| // ---------------- create the input readers --------------------- |
| // in case where a logical input unions multiple physical inputs, create a union reader |
| final int groupSize = this.config.getGroupSize(i); |
| |
| if (groupSize == 1) { |
| // non-union case |
| inputReaders[i] = |
| new MutableRecordReader<>( |
| getEnvironment().getInputGate(currentReaderOffset), |
| getEnvironment().getTaskManagerInfo().getTmpDirectories()); |
| } else if (groupSize > 1) { |
| // union case |
| IndexedInputGate[] readers = new IndexedInputGate[groupSize]; |
| for (int j = 0; j < groupSize; ++j) { |
| readers[j] = getEnvironment().getInputGate(currentReaderOffset + j); |
| } |
| inputReaders[i] = |
| new MutableRecordReader<>( |
| new UnionInputGate(readers), |
| getEnvironment().getTaskManagerInfo().getTmpDirectories()); |
| } else { |
| throw new Exception("Illegal input group size in task configuration: " + groupSize); |
| } |
| |
| currentReaderOffset += groupSize; |
| } |
| this.inputReaders = inputReaders; |
| |
| // final sanity check |
| if (currentReaderOffset != this.config.getNumInputs()) { |
| throw new Exception( |
| "Illegal configuration: Number of input gates and group sizes are not consistent."); |
| } |
| } |
| |
| /** |
| * Creates the record readers for the extra broadcast inputs as configured by {@link |
| * TaskConfig#getNumBroadcastInputs()}. This method requires that the task configuration, the |
| * driver, and the user-code class loader are set. |
| */ |
| protected void initBroadcastInputReaders() throws Exception { |
| final int numBroadcastInputs = this.config.getNumBroadcastInputs(); |
| final MutableReader<?>[] broadcastInputReaders = new MutableReader<?>[numBroadcastInputs]; |
| |
| int currentReaderOffset = config.getNumInputs(); |
| |
| for (int i = 0; i < this.config.getNumBroadcastInputs(); i++) { |
| // ---------------- create the input readers --------------------- |
| // in case where a logical input unions multiple physical inputs, create a union reader |
| final int groupSize = this.config.getBroadcastGroupSize(i); |
| if (groupSize == 1) { |
| // non-union case |
| broadcastInputReaders[i] = |
| new MutableRecordReader<>( |
| getEnvironment().getInputGate(currentReaderOffset), |
| getEnvironment().getTaskManagerInfo().getTmpDirectories()); |
| } else if (groupSize > 1) { |
| // union case |
| IndexedInputGate[] readers = new IndexedInputGate[groupSize]; |
| for (int j = 0; j < groupSize; ++j) { |
| readers[j] = getEnvironment().getInputGate(currentReaderOffset + j); |
| } |
| broadcastInputReaders[i] = |
| new MutableRecordReader<>( |
| new UnionInputGate(readers), |
| getEnvironment().getTaskManagerInfo().getTmpDirectories()); |
| } else { |
| throw new Exception("Illegal input group size in task configuration: " + groupSize); |
| } |
| |
| currentReaderOffset += groupSize; |
| } |
| this.broadcastInputReaders = broadcastInputReaders; |
| } |
| |
| /** Creates all the serializers and comparators. */ |
| protected void initInputsSerializersAndComparators(int numInputs, int numComparators) { |
| this.inputSerializers = new TypeSerializerFactory<?>[numInputs]; |
| this.inputComparators = numComparators > 0 ? new TypeComparator<?>[numComparators] : null; |
| this.inputIterators = new MutableObjectIterator<?>[numInputs]; |
| |
| ClassLoader userCodeClassLoader = getUserCodeClassLoader(); |
| |
| for (int i = 0; i < numInputs; i++) { |
| |
| final TypeSerializerFactory<?> serializerFactory = |
| this.config.getInputSerializer(i, userCodeClassLoader); |
| this.inputSerializers[i] = serializerFactory; |
| |
| this.inputIterators[i] = |
| createInputIterator(this.inputReaders[i], this.inputSerializers[i]); |
| } |
| |
| // ---------------- create the driver's comparators --------------------- |
| for (int i = 0; i < numComparators; i++) { |
| |
| if (this.inputComparators != null) { |
| final TypeComparatorFactory<?> comparatorFactory = |
| this.config.getDriverComparator(i, userCodeClassLoader); |
| this.inputComparators[i] = comparatorFactory.createComparator(); |
| } |
| } |
| } |
| |
| /** Creates all the serializers and iterators for the broadcast inputs. */ |
| protected void initBroadcastInputsSerializers(int numBroadcastInputs) { |
| this.broadcastInputSerializers = new TypeSerializerFactory<?>[numBroadcastInputs]; |
| |
| ClassLoader userCodeClassLoader = getUserCodeClassLoader(); |
| |
| for (int i = 0; i < numBroadcastInputs; i++) { |
| // ---------------- create the serializer first --------------------- |
| final TypeSerializerFactory<?> serializerFactory = |
| this.config.getBroadcastInputSerializer(i, userCodeClassLoader); |
| this.broadcastInputSerializers[i] = serializerFactory; |
| } |
| } |
| |
| /** |
| * NOTE: This method must be invoked after the invocation of {@code #initInputReaders()} and |
| * {@code #initInputSerializersAndComparators(int)}! |
| */ |
| protected void initLocalStrategies(int numInputs) throws Exception { |
| |
| final MemoryManager memMan = getMemoryManager(); |
| final IOManager ioMan = getIOManager(); |
| |
| this.localStrategies = new CloseableInputProvider<?>[numInputs]; |
| this.inputs = new MutableObjectIterator<?>[numInputs]; |
| this.excludeFromReset = new boolean[numInputs]; |
| this.inputIsCached = new boolean[numInputs]; |
| this.inputIsAsyncMaterialized = new boolean[numInputs]; |
| this.materializationMemory = new int[numInputs]; |
| |
| // set up the local strategies first, such that the can work before any temp barrier is |
| // created |
| for (int i = 0; i < numInputs; i++) { |
| initInputLocalStrategy(i); |
| } |
| |
| // we do another loop over the inputs, because we want to instantiate all |
| // sorters, etc before requesting the first input (as this call may block) |
| |
| // we have two types of materialized inputs, and both are replayable (can act as a cache) |
| // The first variant materializes in a different thread and hence |
| // acts as a pipeline breaker. this one should only be there, if a pipeline breaker is |
| // needed. |
| // the second variant spills to the side and will not read unless the result is also |
| // consumed |
| // in a pipelined fashion. |
| this.resettableInputs = new SpillingResettableMutableObjectIterator<?>[numInputs]; |
| this.tempBarriers = new TempBarrier<?>[numInputs]; |
| |
| for (int i = 0; i < numInputs; i++) { |
| final int memoryPages; |
| final boolean async = this.config.isInputAsynchronouslyMaterialized(i); |
| final boolean cached = this.config.isInputCached(i); |
| |
| this.inputIsAsyncMaterialized[i] = async; |
| this.inputIsCached[i] = cached; |
| |
| if (async || cached) { |
| memoryPages = |
| memMan.computeNumberOfPages( |
| this.config.getRelativeInputMaterializationMemory(i)); |
| if (memoryPages <= 0) { |
| throw new Exception( |
| "Input marked as materialized/cached, but no memory for materialization provided."); |
| } |
| this.materializationMemory[i] = memoryPages; |
| } else { |
| memoryPages = 0; |
| } |
| |
| if (async) { |
| @SuppressWarnings({"unchecked", "rawtypes"}) |
| TempBarrier<?> barrier = |
| new TempBarrier( |
| this, |
| getInput(i), |
| this.inputSerializers[i], |
| memMan, |
| ioMan, |
| memoryPages, |
| emptyList()); |
| barrier.startReading(); |
| this.tempBarriers[i] = barrier; |
| this.inputs[i] = null; |
| } else if (cached) { |
| @SuppressWarnings({"unchecked", "rawtypes"}) |
| SpillingResettableMutableObjectIterator<?> iter = |
| new SpillingResettableMutableObjectIterator( |
| getInput(i), |
| this.inputSerializers[i].getSerializer(), |
| getMemoryManager(), |
| getIOManager(), |
| memoryPages, |
| this); |
| this.resettableInputs[i] = iter; |
| this.inputs[i] = iter; |
| } |
| } |
| } |
| |
| protected void resetAllInputs() throws Exception { |
| |
| // first we need to make sure that caches consume remaining data |
| // NOTE: we need to do this before closing the local strategies |
| for (int i = 0; i < this.inputs.length; i++) { |
| |
| if (this.inputIsCached[i] && this.resettableInputs[i] != null) { |
| this.resettableInputs[i].consumeAndCacheRemainingData(); |
| } |
| } |
| |
| // close all local-strategies. they will either get re-initialized, or we have |
| // read them now and their data is cached |
| for (int i = 0; i < this.localStrategies.length; i++) { |
| if (this.localStrategies[i] != null) { |
| this.localStrategies[i].close(); |
| this.localStrategies[i] = null; |
| } |
| } |
| |
| final MemoryManager memMan = getMemoryManager(); |
| final IOManager ioMan = getIOManager(); |
| |
| // reset the caches, or re-run the input local strategy |
| for (int i = 0; i < this.inputs.length; i++) { |
| if (this.excludeFromReset[i]) { |
| if (this.tempBarriers[i] != null) { |
| this.tempBarriers[i].close(); |
| this.tempBarriers[i] = null; |
| } else if (this.resettableInputs[i] != null) { |
| this.resettableInputs[i].close(); |
| this.resettableInputs[i] = null; |
| } |
| } else { |
| // make sure the input is not available directly, but are lazily fetched again |
| this.inputs[i] = null; |
| |
| if (this.inputIsCached[i]) { |
| if (this.tempBarriers[i] != null) { |
| this.inputs[i] = this.tempBarriers[i].getIterator(); |
| } else if (this.resettableInputs[i] != null) { |
| this.resettableInputs[i].reset(); |
| this.inputs[i] = this.resettableInputs[i]; |
| } else { |
| throw new RuntimeException( |
| "Found a resettable input, but no temp barrier and no resettable iterator."); |
| } |
| } else { |
| // close the async barrier if there is one |
| List<MemorySegment> allocated = |
| tempBarriers[i] == null |
| ? emptyList() |
| : tempBarriers[i].closeAndGetLeftoverMemory(); |
| tempBarriers[i] = null; |
| |
| try { |
| initInputLocalStrategy(i); |
| |
| if (this.inputIsAsyncMaterialized[i]) { |
| final int pages = this.materializationMemory[i]; |
| Preconditions.checkState( |
| allocated.size() |
| <= pages); // pages shouldn't change, but some segments |
| // might have been consumed |
| @SuppressWarnings({"unchecked", "rawtypes"}) |
| TempBarrier<?> barrier = |
| new TempBarrier( |
| this, |
| getInput(i), |
| this.inputSerializers[i], |
| memMan, |
| ioMan, |
| pages, |
| allocated); |
| barrier.startReading(); |
| this.tempBarriers[i] = barrier; |
| this.inputs[i] = null; |
| } else { |
| memMan.release(allocated); |
| } |
| } catch (Exception exception) { |
| try { |
| memMan.release(allocated); |
| } catch (Exception releaseException) { |
| exception.addSuppressed(releaseException); |
| } |
| throw exception; |
| } |
| } |
| } |
| } |
| } |
| |
| protected void excludeFromReset(int inputNum) { |
| this.excludeFromReset[inputNum] = true; |
| } |
| |
| private void initInputLocalStrategy(int inputNum) throws Exception { |
| // check if there is already a strategy |
| if (this.localStrategies[inputNum] != null) { |
| throw new IllegalStateException(); |
| } |
| |
| // now set up the local strategy |
| final LocalStrategy localStrategy = this.config.getInputLocalStrategy(inputNum); |
| if (localStrategy != null) { |
| switch (localStrategy) { |
| case NONE: |
| // the input is as it is |
| this.inputs[inputNum] = this.inputIterators[inputNum]; |
| break; |
| case SORT: |
| @SuppressWarnings({"rawtypes", "unchecked"}) |
| Sorter<?> sorter = |
| ExternalSorter.newBuilder( |
| getMemoryManager(), |
| this, |
| this.inputSerializers[inputNum].getSerializer(), |
| getLocalStrategyComparator(inputNum)) |
| .maxNumFileHandles(this.config.getFilehandlesInput(inputNum)) |
| .enableSpilling( |
| getIOManager(), |
| this.config.getSpillingThresholdInput(inputNum)) |
| .memoryFraction(this.config.getRelativeMemoryInput(inputNum)) |
| .objectReuse(this.getExecutionConfig().isObjectReuseEnabled()) |
| .largeRecords(this.getTaskConfig().getUseLargeRecordHandler()) |
| .build((MutableObjectIterator) this.inputIterators[inputNum]); |
| // set the input to null such that it will be lazily fetched from the input |
| // strategy |
| this.inputs[inputNum] = null; |
| this.localStrategies[inputNum] = sorter; |
| break; |
| case COMBININGSORT: |
| // sanity check this special case! |
| // this still breaks a bit of the abstraction! |
| // we should have nested configurations for the local strategies to solve that |
| if (inputNum != 0) { |
| throw new IllegalStateException( |
| "Performing combining sort outside a (group)reduce task!"); |
| } |
| |
| // instantiate ourselves a combiner. we should not use the stub, because the |
| // sort and the |
| // subsequent (group)reduce would otherwise share it multi-threaded |
| final Class<S> userCodeFunctionType = this.driver.getStubType(); |
| if (userCodeFunctionType == null) { |
| throw new IllegalStateException( |
| "Performing combining sort outside a reduce task!"); |
| } |
| final S localStub; |
| try { |
| localStub = initStub(userCodeFunctionType); |
| } catch (Exception e) { |
| throw new RuntimeException( |
| "Initializing the user code and the configuration failed" |
| + (e.getMessage() == null ? "." : ": " + e.getMessage()), |
| e); |
| } |
| |
| if (!(localStub instanceof GroupCombineFunction)) { |
| throw new IllegalStateException( |
| "Performing combining sort outside a reduce task!"); |
| } |
| |
| @SuppressWarnings({"rawtypes", "unchecked"}) |
| Sorter<?> cSorter = |
| ExternalSorter.newBuilder( |
| getMemoryManager(), |
| this, |
| this.inputSerializers[inputNum].getSerializer(), |
| getLocalStrategyComparator(inputNum)) |
| .maxNumFileHandles(this.config.getFilehandlesInput(inputNum)) |
| .withCombiner( |
| (GroupCombineFunction) localStub, |
| this.config.getStubParameters()) |
| .enableSpilling( |
| getIOManager(), |
| this.config.getSpillingThresholdInput(inputNum)) |
| .memoryFraction(this.config.getRelativeMemoryInput(inputNum)) |
| .objectReuse(this.getExecutionConfig().isObjectReuseEnabled()) |
| .largeRecords(this.getTaskConfig().getUseLargeRecordHandler()) |
| .build(this.inputIterators[inputNum]); |
| |
| // set the input to null such that it will be lazily fetched from the input |
| // strategy |
| this.inputs[inputNum] = null; |
| this.localStrategies[inputNum] = cSorter; |
| break; |
| default: |
| throw new Exception( |
| "Unrecognized local strategy provided: " + localStrategy.name()); |
| } |
| } else { |
| // no local strategy in the config |
| this.inputs[inputNum] = this.inputIterators[inputNum]; |
| } |
| } |
| |
| private <T> TypeComparator<T> getLocalStrategyComparator(int inputNum) throws Exception { |
| TypeComparatorFactory<T> compFact = |
| this.config.getInputComparator(inputNum, getUserCodeClassLoader()); |
| if (compFact == null) { |
| throw new Exception( |
| "Missing comparator factory for local strategy on input " + inputNum); |
| } |
| return compFact.createComparator(); |
| } |
| |
| protected MutableObjectIterator<?> createInputIterator( |
| MutableReader<?> inputReader, TypeSerializerFactory<?> serializerFactory) { |
| @SuppressWarnings("unchecked") |
| MutableReader<DeserializationDelegate<?>> reader = |
| (MutableReader<DeserializationDelegate<?>>) inputReader; |
| @SuppressWarnings({"unchecked", "rawtypes"}) |
| final MutableObjectIterator<?> iter = |
| new ReaderIterator(reader, serializerFactory.getSerializer()); |
| return iter; |
| } |
| |
| protected int getNumTaskInputs() { |
| return this.driver.getNumberOfInputs(); |
| } |
| |
| /** |
| * Creates a writer for each output. Creates an OutputCollector which forwards its input to all |
| * writers. The output collector applies the configured shipping strategies for each writer. |
| */ |
| protected void initOutputs() throws Exception { |
| this.chainedTasks = new ArrayList<>(); |
| this.eventualOutputs = new ArrayList<>(); |
| |
| this.accumulatorMap = getEnvironment().getAccumulatorRegistry().getUserMap(); |
| |
| this.output = |
| initOutputs( |
| this, |
| getEnvironment().getUserCodeClassLoader(), |
| this.config, |
| this.chainedTasks, |
| this.eventualOutputs, |
| this.getExecutionConfig(), |
| this.accumulatorMap); |
| } |
| |
| public DistributedRuntimeUDFContext createRuntimeContext(OperatorMetricGroup metrics) { |
| Environment env = getEnvironment(); |
| |
| return new DistributedRuntimeUDFContext( |
| env.getJobInfo(), |
| env.getTaskInfo(), |
| env.getUserCodeClassLoader(), |
| getExecutionConfig(), |
| env.getDistributedCacheEntries(), |
| this.accumulatorMap, |
| metrics, |
| env.getExternalResourceInfoProvider()); |
| } |
| |
| // -------------------------------------------------------------------------------------------- |
| // Task Context Signature |
| // ------------------------------------------------------------------------------------------- |
| |
| @Override |
| public TaskConfig getTaskConfig() { |
| return this.config; |
| } |
| |
| @Override |
| public TaskManagerRuntimeInfo getTaskManagerInfo() { |
| return getEnvironment().getTaskManagerInfo(); |
| } |
| |
| @Override |
| public MemoryManager getMemoryManager() { |
| return getEnvironment().getMemoryManager(); |
| } |
| |
| @Override |
| public IOManager getIOManager() { |
| return getEnvironment().getIOManager(); |
| } |
| |
| @Override |
| public S getStub() { |
| return this.stub; |
| } |
| |
| @Override |
| public Collector<OT> getOutputCollector() { |
| return this.output; |
| } |
| |
| @Override |
| public AbstractInvokable getContainingTask() { |
| return this; |
| } |
| |
| @Override |
| public String formatLogString(String message) { |
| return constructLogString(message, getEnvironment().getTaskInfo().getTaskName(), this); |
| } |
| |
| @Override |
| public OperatorMetricGroup getMetricGroup() { |
| return metrics; |
| } |
| |
| @Override |
| public <X> MutableObjectIterator<X> getInput(int index) { |
| if (index < 0 || index > this.driver.getNumberOfInputs()) { |
| throw new IndexOutOfBoundsException(); |
| } |
| |
| // check for lazy assignment from input strategies |
| if (this.inputs[index] != null) { |
| @SuppressWarnings("unchecked") |
| MutableObjectIterator<X> in = (MutableObjectIterator<X>) this.inputs[index]; |
| return in; |
| } else { |
| final MutableObjectIterator<X> in; |
| try { |
| if (this.tempBarriers[index] != null) { |
| @SuppressWarnings("unchecked") |
| MutableObjectIterator<X> iter = |
| (MutableObjectIterator<X>) this.tempBarriers[index].getIterator(); |
| in = iter; |
| } else if (this.localStrategies[index] != null) { |
| @SuppressWarnings("unchecked") |
| MutableObjectIterator<X> iter = |
| (MutableObjectIterator<X>) this.localStrategies[index].getIterator(); |
| in = iter; |
| } else { |
| throw new RuntimeException( |
| "Bug: null input iterator, null temp barrier, and null local strategy."); |
| } |
| this.inputs[index] = in; |
| return in; |
| } catch (InterruptedException iex) { |
| throw new RuntimeException( |
| "Interrupted while waiting for input " + index + " to become available."); |
| } catch (IOException ioex) { |
| throw new RuntimeException( |
| "An I/O Exception occurred while obtaining input " + index + "."); |
| } |
| } |
| } |
| |
| @Override |
| public <X> TypeSerializerFactory<X> getInputSerializer(int index) { |
| if (index < 0 || index >= this.driver.getNumberOfInputs()) { |
| throw new IndexOutOfBoundsException(); |
| } |
| |
| @SuppressWarnings("unchecked") |
| final TypeSerializerFactory<X> serializerFactory = |
| (TypeSerializerFactory<X>) this.inputSerializers[index]; |
| return serializerFactory; |
| } |
| |
| @Override |
| public <X> TypeComparator<X> getDriverComparator(int index) { |
| if (this.inputComparators == null) { |
| throw new IllegalStateException("Comparators have not been created!"); |
| } else if (index < 0 || index >= this.driver.getNumberOfDriverComparators()) { |
| throw new IndexOutOfBoundsException(); |
| } |
| |
| @SuppressWarnings("unchecked") |
| final TypeComparator<X> comparator = (TypeComparator<X>) this.inputComparators[index]; |
| return comparator; |
| } |
| |
| // ============================================================================================ |
| // Static Utilities |
| // |
| // Utilities are consolidated here to ensure a uniform way of running, |
| // logging, exception handling, and error messages. |
| // ============================================================================================ |
| |
| // -------------------------------------------------------------------------------------------- |
| // Logging |
| // -------------------------------------------------------------------------------------------- |
| /** |
| * Utility function that composes a string for logging purposes. The string includes the given |
| * message, the given name of the task and the index in its subtask group as well as the number |
| * of instances that exist in its subtask group. |
| * |
| * @param message The main message for the log. |
| * @param taskName The name of the task. |
| * @param parent The task that contains the code producing the message. |
| * @return The string for logging. |
| */ |
| public static String constructLogString( |
| String message, String taskName, AbstractInvokable parent) { |
| return message |
| + ": " |
| + taskName |
| + " (" |
| + (parent.getEnvironment().getTaskInfo().getIndexOfThisSubtask() + 1) |
| + '/' |
| + parent.getEnvironment().getTaskInfo().getNumberOfParallelSubtasks() |
| + ')'; |
| } |
| |
| /** |
| * Prints an error message and throws the given exception. If the exception is of the type |
| * {@link ExceptionInChainedStubException} then the chain of contained exceptions is followed |
| * until an exception of a different type is found. |
| * |
| * @param ex The exception to be thrown. |
| * @param parent The parent task, whose information is included in the log message. |
| * @throws Exception Always thrown. |
| */ |
| public static void logAndThrowException(Exception ex, AbstractInvokable parent) |
| throws Exception { |
| String taskName; |
| if (ex instanceof ExceptionInChainedStubException) { |
| do { |
| ExceptionInChainedStubException cex = (ExceptionInChainedStubException) ex; |
| taskName = cex.getTaskName(); |
| ex = cex.getWrappedException(); |
| } while (ex instanceof ExceptionInChainedStubException); |
| } else { |
| taskName = parent.getEnvironment().getTaskInfo().getTaskName(); |
| } |
| |
| if (LOG.isErrorEnabled()) { |
| LOG.error(constructLogString("Error in task code", taskName, parent), ex); |
| } |
| |
| throw ex; |
| } |
| |
| // -------------------------------------------------------------------------------------------- |
| // Result Shipping and Chained Tasks |
| // -------------------------------------------------------------------------------------------- |
| |
| /** |
| * Creates the {@link Collector} for the given task, as described by the given configuration. |
| * The output collector contains the writers that forward the data to the different tasks that |
| * the given task is connected to. Each writer applies the partitioning as described in the |
| * configuration. |
| * |
| * @param task The task that the output collector is created for. |
| * @param config The configuration describing the output shipping strategies. |
| * @param cl The classloader used to load user defined types. |
| * @param eventualOutputs The output writers that this task forwards to the next task for each |
| * output. |
| * @param outputOffset The offset to start to get the writers for the outputs |
| * @param numOutputs The number of outputs described in the configuration. |
| * @return The OutputCollector that data produced in this task is submitted to. |
| */ |
| public static <T> Collector<T> getOutputCollector( |
| AbstractInvokable task, |
| TaskConfig config, |
| ClassLoader cl, |
| List<RecordWriter<?>> eventualOutputs, |
| int outputOffset, |
| int numOutputs) |
| throws Exception { |
| if (numOutputs == 0) { |
| return null; |
| } |
| |
| // get the factory for the serializer |
| final TypeSerializerFactory<T> serializerFactory = config.getOutputSerializer(cl); |
| final List<RecordWriter<SerializationDelegate<T>>> writers = new ArrayList<>(numOutputs); |
| |
| // create a writer for each output |
| for (int i = 0; i < numOutputs; i++) { |
| // create the OutputEmitter from output ship strategy |
| final ShipStrategyType strategy = config.getOutputShipStrategy(i); |
| final int indexInSubtaskGroup = task.getIndexInSubtaskGroup(); |
| final TypeComparatorFactory<T> compFactory = config.getOutputComparator(i, cl); |
| |
| final ChannelSelector<SerializationDelegate<T>> oe; |
| if (compFactory == null) { |
| oe = new OutputEmitter<>(strategy, indexInSubtaskGroup); |
| } else { |
| final DataDistribution dataDist = config.getOutputDataDistribution(i, cl); |
| final Partitioner<?> partitioner = config.getOutputPartitioner(i, cl); |
| |
| final TypeComparator<T> comparator = compFactory.createComparator(); |
| oe = |
| new OutputEmitter<>( |
| strategy, indexInSubtaskGroup, comparator, partitioner, dataDist); |
| } |
| |
| final RecordWriter<SerializationDelegate<T>> recordWriter = |
| new RecordWriterBuilder() |
| .setChannelSelector(oe) |
| .setTaskName( |
| task.getEnvironment().getTaskInfo().getTaskNameWithSubtasks()) |
| .build(task.getEnvironment().getWriter(outputOffset + i)); |
| |
| recordWriter.setMetricGroup(task.getEnvironment().getMetricGroup().getIOMetricGroup()); |
| |
| writers.add(recordWriter); |
| } |
| if (eventualOutputs != null) { |
| eventualOutputs.addAll(writers); |
| } |
| return new OutputCollector<>(writers, serializerFactory.getSerializer()); |
| } |
| |
| /** |
| * Creates a writer for each output. Creates an OutputCollector which forwards its input to all |
| * writers. The output collector applies the configured shipping strategy. |
| */ |
| @SuppressWarnings("unchecked") |
| public static <T> Collector<T> initOutputs( |
| AbstractInvokable containingTask, |
| UserCodeClassLoader cl, |
| TaskConfig config, |
| List<ChainedDriver<?, ?>> chainedTasksTarget, |
| List<RecordWriter<?>> eventualOutputs, |
| ExecutionConfig executionConfig, |
| Map<String, Accumulator<?, ?>> accumulatorMap) |
| throws Exception { |
| final int numOutputs = config.getNumOutputs(); |
| |
| // check whether we got any chained tasks |
| final int numChained = config.getNumberOfChainedStubs(); |
| if (numChained > 0) { |
| // got chained stubs. that means that this one may only have a single forward connection |
| if (numOutputs != 1 || config.getOutputShipStrategy(0) != ShipStrategyType.FORWARD) { |
| throw new RuntimeException( |
| "Plan Generation Bug: Found a chained stub that is not connected via an only forward connection."); |
| } |
| |
| // instantiate each task |
| @SuppressWarnings("rawtypes") |
| Collector previous = null; |
| for (int i = numChained - 1; i >= 0; --i) { |
| // get the task first |
| final ChainedDriver<?, ?> ct; |
| try { |
| Class<? extends ChainedDriver<?, ?>> ctc = config.getChainedTask(i); |
| ct = ctc.newInstance(); |
| } catch (Exception ex) { |
| throw new RuntimeException("Could not instantiate chained task driver.", ex); |
| } |
| |
| // get the configuration for the task |
| final TaskConfig chainedStubConf = config.getChainedStubConfig(i); |
| final String taskName = config.getChainedTaskName(i); |
| |
| if (i == numChained - 1) { |
| // last in chain, instantiate the output collector for this task |
| previous = |
| getOutputCollector( |
| containingTask, |
| chainedStubConf, |
| cl.asClassLoader(), |
| eventualOutputs, |
| 0, |
| chainedStubConf.getNumOutputs()); |
| } |
| |
| ct.setup( |
| chainedStubConf, |
| taskName, |
| previous, |
| containingTask, |
| cl, |
| executionConfig, |
| accumulatorMap); |
| chainedTasksTarget.add(0, ct); |
| |
| if (i == numChained - 1) { |
| ct.getIOMetrics().reuseOutputMetricsForTask(); |
| } |
| |
| previous = ct; |
| } |
| // the collector of the first in the chain is the collector for the task |
| return (Collector<T>) previous; |
| } |
| // else |
| |
| // instantiate the output collector the default way from this configuration |
| return getOutputCollector( |
| containingTask, config, cl.asClassLoader(), eventualOutputs, 0, numOutputs); |
| } |
| |
| // -------------------------------------------------------------------------------------------- |
| // User Code LifeCycle |
| // -------------------------------------------------------------------------------------------- |
| |
| /** |
| * Opens the given stub using its {@link |
| * org.apache.flink.api.common.functions.RichFunction#open(OpenContext)} method. If the open |
| * call produces an exception, a new exception with a standard error message is created, using |
| * the encountered exception as its cause. |
| * |
| * @param stub The user code instance to be opened. |
| * @param parameters The parameters supplied to the user code. |
| * @throws Exception Thrown, if the user code's open method produces an exception. |
| */ |
| public static void openUserCode(Function stub, Configuration parameters) throws Exception { |
| try { |
| FunctionUtils.openFunction(stub, DefaultOpenContext.INSTANCE); |
| } catch (Throwable t) { |
| throw new Exception( |
| "The user defined 'open(OpenContext)' method in " |
| + stub.getClass().toString() |
| + " caused an exception: " |
| + t.getMessage(), |
| t); |
| } |
| } |
| |
| /** |
| * Closes the given stub using its {@link |
| * org.apache.flink.api.common.functions.RichFunction#close()} method. If the close call |
| * produces an exception, a new exception with a standard error message is created, using the |
| * encountered exception as its cause. |
| * |
| * @param stub The user code instance to be closed. |
| * @throws Exception Thrown, if the user code's close method produces an exception. |
| */ |
| public static void closeUserCode(Function stub) throws Exception { |
| try { |
| FunctionUtils.closeFunction(stub); |
| } catch (Throwable t) { |
| throw new Exception( |
| "The user defined 'close()' method caused an exception: " + t.getMessage(), t); |
| } |
| } |
| |
| // -------------------------------------------------------------------------------------------- |
| // Chained Task LifeCycle |
| // -------------------------------------------------------------------------------------------- |
| |
| /** |
| * Opens all chained tasks, in the order as they are stored in the array. The opening process |
| * creates a standardized log info message. |
| * |
| * @param tasks The tasks to be opened. |
| * @param parent The parent task, used to obtain parameters to include in the log message. |
| * @throws Exception Thrown, if the opening encounters an exception. |
| */ |
| public static void openChainedTasks(List<ChainedDriver<?, ?>> tasks, AbstractInvokable parent) |
| throws Exception { |
| // start all chained tasks |
| for (ChainedDriver<?, ?> task : tasks) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug(constructLogString("Start task code", task.getTaskName(), parent)); |
| } |
| task.openTask(); |
| } |
| } |
| |
| /** |
| * Closes all chained tasks, in the order as they are stored in the array. The closing process |
| * creates a standardized log info message. |
| * |
| * @param tasks The tasks to be closed. |
| * @param parent The parent task, used to obtain parameters to include in the log message. |
| * @throws Exception Thrown, if the closing encounters an exception. |
| */ |
| public static void closeChainedTasks(List<ChainedDriver<?, ?>> tasks, AbstractInvokable parent) |
| throws Exception { |
| for (ChainedDriver<?, ?> task : tasks) { |
| task.closeTask(); |
| |
| if (LOG.isDebugEnabled()) { |
| LOG.debug(constructLogString("Finished task code", task.getTaskName(), parent)); |
| } |
| } |
| } |
| |
| /** |
| * Cancels all tasks via their {@link ChainedDriver#cancelTask()} method. Any occurring |
| * exception and error is suppressed, such that the canceling method of every task is invoked in |
| * all cases. |
| * |
| * @param tasks The tasks to be canceled. |
| */ |
| public static void cancelChainedTasks(List<ChainedDriver<?, ?>> tasks) { |
| for (ChainedDriver<?, ?> task : tasks) { |
| try { |
| task.cancelTask(); |
| } catch (Throwable t) { |
| // do nothing |
| } |
| } |
| } |
| |
| // -------------------------------------------------------------------------------------------- |
| // Miscellaneous Utilities |
| // -------------------------------------------------------------------------------------------- |
| |
| /** |
| * Instantiates a user code class from is definition in the task configuration. The class is |
| * instantiated without arguments using the null-ary constructor. Instantiation will fail if |
| * this constructor does not exist or is not public. |
| * |
| * @param <T> The generic type of the user code class. |
| * @param config The task configuration containing the class description. |
| * @param cl The class loader to be used to load the class. |
| * @param superClass The super class that the user code class extends or implements, for type |
| * checking. |
| * @return An instance of the user code class. |
| */ |
| public static <T> T instantiateUserCode( |
| TaskConfig config, ClassLoader cl, Class<? super T> superClass) { |
| try { |
| T stub = config.<T>getStubWrapper(cl).getUserCodeObject(superClass, cl); |
| // check if the class is a subclass, if the check is required |
| if (superClass != null && !superClass.isAssignableFrom(stub.getClass())) { |
| throw new RuntimeException( |
| "The class '" |
| + stub.getClass().getName() |
| + "' is not a subclass of '" |
| + superClass.getName() |
| + "' as is required."); |
| } |
| return stub; |
| } catch (ClassCastException ccex) { |
| throw new RuntimeException( |
| "The UDF class is not a proper subclass of " + superClass.getName(), ccex); |
| } |
| } |
| |
| private static int[] asArray(List<Integer> list) { |
| int[] a = new int[list.size()]; |
| |
| int i = 0; |
| for (int val : list) { |
| a[i++] = val; |
| } |
| return a; |
| } |
| |
| public static void clearWriters(List<RecordWriter<?>> writers) { |
| for (RecordWriter<?> writer : writers) { |
| writer.close(); |
| } |
| } |
| |
| public static void clearReaders(MutableReader<?>[] readers) { |
| for (MutableReader<?> reader : readers) { |
| reader.clearBuffers(); |
| } |
| } |
| } |
