[ 
https://issues.apache.org/jira/browse/APEXMALHAR-2278?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Ananth closed APEXMALHAR-2278.
------------------------------

> Implement Kudu Output Operator for non-transactional streams
> ------------------------------------------------------------
>
>                 Key: APEXMALHAR-2278
>                 URL: https://issues.apache.org/jira/browse/APEXMALHAR-2278
>             Project: Apache Apex Malhar
>          Issue Type: New Feature
>          Components: adapters database
>            Reporter: Ananth
>            Assignee: Ananth
>             Fix For: 3.8.0
>
>
> Here are some benefits of integrating Kudu and Apex:
>     Kudu is just declared 1.0 and has just been declared production ready.
>     Kudu as a store might a good a fit for many architectures in the years to 
> come because of its capabilities to provide mutability of data ( unlike HDFS 
> ) and optimized storage formats for low latency scans.
>     It seems to also withstand high-throughput write patterns which makes it 
> a stable sink for Apex workflows which operate at very high volumes. 
> [Design] 
> 1. The operator would be an AbstractOperator and would allow the concrete 
> implementations to set a few behavioral aspects of the operator. 
> 2. The following are the major phases of the operator:
>     During activate() phase of the operator : Establish a connection to the 
> cluster and get the metadata about the table that is being used as the sink.
>     During setup() phase of the operator: Fetch the current window 
> information and use it decide if we are recovering from a failure mode. (See 
> point 8 below )
>     During process() of Input port : Inspect the incoming ExecutionContext ( 
> see below ) tuple and perform one of the  operations ( 
> Insert/Update/Delete/Upsert) 
> 3. The following parameters are tunable while establishing a Kudu connection:
>     Table name, Boss worker threads, Worker threads, Socket read time outs 
> and External Consistency mode.
> 4. The user need not specify any schema outright. The pojo fields are 
> automatically mapped to the table column names as identified in the schema 
> parse in the activate phase. 
> 5. Allow the concrete implementation of the operator to override the Pojo 
> field name to the table schema column name. This would allow flexibility in 
> use cases like table schema column names are not compatible with java bean 
> frameworks or in situations when column names cant be controlled as POJO is 
> coming from an upstream operator.
> 6. The input tuple that is to be supplied to this operator is of type "Kudu 
> Execution Context". This tuple encompasses the actual Pojo that is going to 
> be persisted to the Kudu store. Additionally it allows the upstream operator 
> to specify the operation that needs to be performed. One of the following 
> operations is permitted as part of the context : Insert, Upsert, Update and 
> delete on the Pojo that is acting as the payload in the Execution Context.
> 7. The concrete implementation of the operator would allow the user to 
> specify the actual POJO class definition that would be used to the write to 
> the table. The execution context would contain this POJO as well as the 
> metadata that defines the behavior of the processing that needs to be done on 
> that tuple.
> 8. The operator would allow for a special case of execution mode for the 
> first window that is being processed as the operator gets activated. There 
> are two modes for the first window of processing of the operator : 
>       a. Safe Mode :  Safe mode is the "happy path execution" as in no extra 
> processing is required to perform the Kudu mutation.
>       b. Reconciling Mode:  There is an additional function that would be 
> called to see if the user would like the tuple to be used for mutation. This 
> mode is automatically set when OperatorContext.ACTIVATION_WINDOW_ID != 
> Stateless.WINDOW_ID   during the first window of processing by the operator. 
> This feature is deemed to be useful when an operator is recovering from a 
> crash instance of the application and we do not want to perform multiple 
> mutations of the same tuple given ATLEAST_ONCE is the default semantics.  
> 9. The operator is a stateless operator. 
> 10. The operator would generate the following autometrics :
>       a. Counts  of Inserts, Upserts, Deletes and Updates (separate counters 
> for each mutation) for a given window
>       b. Bytes written in a given window
>       c. Write RPCs in the given window
>       d. Total RPC errors in this window
>        e. All of the above metrics for the operator for its entire lifetime 
> of the operator. 



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

Reply via email to