Skip to content

udf-implementation.md

Latest commit

 

History

History
78 lines (68 loc) · 4.2 KB

udf-implementation.md

File metadata and controls

78 lines (68 loc) · 4.2 KB

UDF Implementation

Library architecture

SDP library is implementing IUDFAgent for communicating with Kapacitor via Kapacitor's UDF RPC protocol. UDFs use socket based approach to communicate with Kapacitor, but UDFAgent is ready to work with child process based approach if needed.

We use uvw library, which is libuv wrapper, in order to provide asynchronous I/O. It also allows to write code in event-based approach.

Every UDF requires separate RequestHandler interface implementation -- it handles incoming RPC calls from Kapacitor and sends responses back via IUDFAgent.

As long as we are using Apache Arrow library for data processing we need to store, handle data and convert it between Kapacitor's points and Arrow's RecordBatches formats. This functionality is provided by corresponding objects:

Therefore, the most important part of every UDF is a RecordBatchHandler that is doing all useful work. See the full list of currently implemented RecordBatchHandlers.

Writing your own UDF

If you want to write your own UDF using SDP library you could follow the basic steps. As example, you can refer to streamAggregateUDF's and batchAggregateUDF's RequestHandeler implementations (items 3, 4, 6) and to the udf_agent_client_factory.h file (item 7).

  1. Check out if your UDF is expressed through one of RecordBatchHandlers or their composition. If it's not then implement new RecordBatchHandler that has minimal missing functionality for your UDF.
  2. Decide which type does your UDF have: does it consume stream or batch data? Choose appropriate base class for your UDF's RequestHandler: for wants-batch you should probably use BatchRecordBatchRequestHandlerBase, for wants-stream one of StreamRecordBatchRequestHandlerBase and TimerRecordBatchRequestHandlerBase should be suitable for you.
  3. Decide which parameters your UDF should have in terms of Kapacitor RPC Protocol. Implement RequestHandler::info method according to UDF's parameters and type.
  4. Create instrument for parsing UDF's parameters from init RPC message.
  5. If you need additional pre-processing for converting Kapacitor's points to Arrow's RecordBatches, you can implement your own PointsConverter decorator.
  6. Implement RequestHandler::init method. There you should:
    • use your parameters parser from item 3 to parse agent::InitMessage RPC message;
    • create RecordBatchHandler from item 1 according to your UDF's functionality;
    • create PointsConverter with possibly implemented decorator from item 5;
    • create PointsStorage injecting just created RecordBatchHandler and PointsConverter and set it using setPointsStorage protected method.
  7. Implement UnixSocketClientFactory that will be creating AgentClients with UDFAgent and just implemented RequestHandler in it. Use this factory to generate connections to UnixSocketServer.