Operators in Clara are containerized applications that perform a function on given data. Operators are “units” in a pipeline, which itself is a dependency tree of operator. A dependency between operators is created by declaring that the input of one or more operators is the output of another.
At the heart of any operator declaration is its container declaration, specifically the image it uses to perform the function. The image property defines the name of the container image Clara will fetch, deploy, and execute as part of the pipeline execution.
As part of the container declaration, tag and command can also be declared. The tag property is used to determine which version of a given image Clara Deploy SDK should select from its container image repository. The command property is an array (you can tell by the
] characters), and allows a set of strings to be passed to the container and used as the command it executes upon starting.
0.4.0and below operators may not specify type information about their inputs and outputs, and as a result operators may only exchange data through disk I/O (volumes mounted as local directories in the container).
0.5.0and above operators must specify type information for inputs and outputs which allows data exchange between operators through both shared memory and disk I/O. Type information allows for pre-runtime validation of data compatibility between operators, the automatic selection of the channel of communication between operators (shared memory or disk), and scoping and access of data exchanged through shared memory. NOTE: This functionality is only available with Clara orchestration (not Argo).
When operators exchange data via disk I/O, data is loaded in a volume inside the container image where it is accessible by the user’s application. To the application these mounted volumes appear as locally accessible directories. Alternatively, when type information is made available in operator inputs and outputs, the Clara orchestrator has the ability to choose either shared memory or a mounted volume to access the data. Before going into the details of what types are assigned to shared memory or disk, let us first review the types supported for inputs and outputs in the operator definition.
8.4. Clara Types (pipeline
0.5.0 and above)¶
Clara I/O types can be divided into two, Clara primitives and Clara structures, albeit implementationally Clara primitives are only shirt-hand representations of structures (more on this below).
There are 3 Clara structures at present:
arrayis used for shared memory I/O; it indicates to the Clara orchestrator that the input (or output) should be allocated into shared memory.
element-typewhich determines the type of each element in the array and must be a Clara primitive (see table below),
- and a
shapewhich determines the number of dimensions and size of the array.
stringis syntactic sugar for
array<uint8>[-1]which is a short-hand notation for “a dynamic array of bytes” (
-1is used to indicate arrays resizable at runtime).
streamis used for disk I/O; it indicates to the Clara orchestrator that the input (or output) is mounted as a local directory from disk.
element-typeattribute, which is free-form. The user may enter anything as an
element-typethat describes the meaning of the input or output when the input is a
stream, but if this is tied to be an input from another operator to to output to another operator the
element-types must match.
pathattribute, which is a locally mounted volume with the data to be read and/or to be written.
Clara primitives are also syntactic sugar for various
array representations, all of which use shared memory allocations.
|Clara Primitive||Logical Interpretation||
||8-bit unsigned integer||
||16-bit unsigned integer||
||32-bit unsigned integer||
||64-bit unsigned integer||
||8-bit unsigned integer||
||16-bit signed integer||
||32-bit signed integer||
||64-bit signed integer||
8.5. The Anatomy of an Operator¶
So far we’ve covered the concepts of typed and untyped operator inputs and outputs, and Clara types and their properties. Now, let us explore the structure of an operator declaration. Clara allows operators to be bound (declared within a pipeline) or unbound (declared independently and inported into a pipeline at a later point).
8.5.1. Disk-based I/O¶
Let us examine the unbound operator below.
name: i-copy-data container: image: ubuntu tag: 18.04 command: ['sh', '-c', 'cp', '-r', '/input', '/output' ] input: - path: /input type: stream # type `stream` requires `path` element-type: dicom # anything is acceptable if type is stream, however, during pipeline validation the output `element-type` of the upstream operator must match this one output: - name: my-output path: /output type: stream element-type: dicom # anything is acceptable if type is stream, however, during pipeline validation the input type of all downstream operator must match this one
Let us examine the operator.
/inputfolder is a Clara-managed folder made available to the operator’s container because of the operator’s input declaration. The path property will map the input payload to the operator container’s
/inputfolder as a read-only mount. The container execution can then make use of the contents of the folder in any way its author sees fit.
- Next, notice that the
/outputfolder is declared as part of the operator’s output property. This means that it too is Clara-managed storage, and the folder with be mapped as a writable mount. This path can be loaded by other operators to read from using its
Note that, if the pipeline defintion is using
0.4.0 and below, the fields
element-type will be ignored. These are mandatory only for pipeline
0.5.0 and above. For clarity, when using
0.4.0 the operator could be declared as
name: i-copy-data container: image: ubuntu tag: 18.04 command: ['sh', '-c', 'cp', '-r', '/input', '/output' ] input: - path: /input output: - name: my-output path: /output
8.6. Argo vs. Clara Orchestration¶
Operators development follows a different paradigm based on which orchestration engine is used.
- When using Argo, operators can be as simple as a Bash or Python script; there are no specific libraries that need to be used to make things work. However, Argo will start operators lazily, incurring an overhead of approximately 2 seconds per operator. This is because Argo assigns one operator per Kubernetes pod, and the cost of starting up a pod is about 2 seconds depending on the system it is run.
- Clara orchestration requires the use of the Clara libraries, currently supporting C++ or Python. The Clara orchestration engine loads pipelines lazily, but operators greedily, thereby incurring a cost of about 2 seconds to start a pipeline. This is because the Clara orchestrator starts all operators in the same Kubernetes pod. For these reasons, it is recommended that Argo orchestration be used during operator development and integration testing, but Clara orchestration be used for more performance-demanding applications.
The following examples show minimal operators following each orchestrator’s programming paradigm.
Argo orchestrator-compatible operator.
import glob input_path = '/input' def execute(): printf(glob.glob(input_path + '/*.dcm')) if __name__ == "__main__": execute()
It is apparent that some information when using Argo has to be assumed, as is for instance the
/input path, which means that if the input path is changed at the level of the operator description (YAML) there is no utility allowing to retrieve this update from within the code.
- It is possible to retrieve this information for environment variables passed by the Clara platform, for instance by querying
os.getenv(NVIDIA_CLARA_INPUTPATHS), however, this approach is not recommended since the developer’s code might not be compatible with any updates made to the structure of
Clara orchestrator-compatible operator.
import glob from clara import Driver, Payload, Error from clara.logging import perf_logger def execute(driver: Driver, payload: Payload): input_path = None for entry in payload.input_entries: if entry.name == 'payload': input_path = entry.path if input_path: printf(glob.glob(input_path + "/*.dcm")) else: raise Error("No input path provided.") if __name__ == "__main__": perf_logger.processing_started("Started") driver = Driver(execute_handler=execute) # instantiate Clara orchestration driver with `execute` as callback driver.start() # start execution thread driver.wait_for_completion() # block until callback completes perf_logger.processing_ended("Ended")
When using Clara orchestration, the user code is wrapped by the Clara Pipeline Driver (CPDriver) code. All operators are started at the same time in the same pod, but all operators with an upstream I/O dependency (whether shared memory or disk I/O) are blocked until all upstream operators complete. The callback function provided in
execute_handler is triggered once the all locks the operator is waiting on are released.
There are two parameters in the callback function:
driverholds CPDriver metadata as dictated by Clara platform. For more information on the
driverobject please refer to the CPDriver documentation.
payloadprovides all payload information related to both disk-based and shared memory inputs and outputs. For more information on the
payloadobject please refer to the Payloads documentation