KinesisConnectorRecordProcessor about amazon-kinesis-connectors

[BenASchwartz]

The ITransformer, IBuffer, IFilter, and IEmitter interfaces are designed to allow for custom record processing methods.

If the records are different data types, one approach is to wrap all records in a class that can identify the record type before putting them into the Kinesis stream. In the ITransformer, the records will be transformed to this superclass and passed through the IFilter. The IFilter can filter records based on the record type field in the class. Next, the IBuffer can store these records and even perform aggregate operations on them as they are stored. Finally, the IEmitter can take your implementation of IBuffer and perform separate operations on each of the record types.

A simple example would be if you have two types of records coming through your Kinesis stream: A and B. We want to store records of type A in a DynamoDB table TableA and store the records of type B in DynamoDB table TableB. First, write an implementation of IBuffer that stores A and B records in two separate buffers. Then override DynamoDBEmitter's emit method to access the separate buffers and perform a BatchRequest on both tables.

[harlow]

@BenASchwartz thanks for the detailed write up.

If the records are different data types, one approach is to wrap all records in a class that can identify the record type before putting them into the Kinesis stream. In the ITransformer, the records will be transformed to this superclass and passed through the IFilter.

Can you give an example of what that JSON would look like?

Would you suggest a node in the JSON structure:

{
   "employee" : { 
      "name" : "Bob Smith",
      "id": 1,
      "phone" : "613-555-5555"
    }
}

Or having a structure with type and attributes

{
  "type": "employee",
  "attributes" : { 
     "name" : "Bob Smith",
     "id": 1,
     "phone" : "613-555-5555"
   }
}

Then in the Java pipeline we could do something like this?

@Override
public ITransformer<KinesisMessageModel, byte[]> getTransformer(KinesisConnectorConfiguration configuration) {
  // somehow grab message type
  // or is there a way to infer the Class from the JSON directly?
  val className = jsonNode.get("type").asText()
  return Class.forName(className)
}

Any help would be appreciated.

[BenASchwartz]

Every incoming record must have the same wrapping class so that it can be processed by the same ITransformer. For example if you have two types of records: employees and customers you could wrap them in a person object or even something more general. Since the ITransformer implementation must be aware of the data types, it makes sense to serialize the objects as JSON blobs directly from the class.

{
    "person" : {
        "customer" : {
           "name" : "Bob Smith",
           "customerId" : 1,
           "phone" : "613-555-5555"
        }
    }  
}

{
    "person" : {
        "employee" : {
           "name" : "Jane Johnson",
           "employeeId" : 1,
           "phone" : "613-555-5555"
        }
    }  
}

That way every message can be deserialized as a Person object. The Person class could look like this:

public class Person{
  private Employee emp;
  private Customer cust;

  // Getters and Setters for emp and cust
  public Employee getEmployee(){ return emp; }
  public Customer getCustomer(){ return cust; }
  public void setEmployee(Employee emp) { this.emp = emp; }
  public void setCustomer(Customer cust) { this.cust = cust; }

  // useful for processing
  public boolean isEmployee() { return emp != null; }
  public boolean isCustomer() { return cust != null; }
}

And you can process the messages like this:
in ITransformer:

public Person toClass(Record record){
    Person person = deserialize(record); // deserialize can be implemented with any JSON library
    if(person.isEmployee()){
       // Do something for employees
    }
    if(person.isCustomer()){
      // Do something for customers
    }
}

in IEmitter:

List<Person> emit(UnmodifiableBuffer<Person> buffer) throws IOException{
   for(Person p: buffer){
     if(person.isEmployee()){
       // Do something for employees
    }
    if(person.isCustomer()){
      // Do something for customers
    }
  }
}

[harlow]

This is great. 👍 thanks for explaining in such detail.

[Diamonds0a]

Hi, sorry for posting on an old thread.

I am looking through the redshiftbasic sample, and I wanted to ask how this applies to that sample.

Going with what you described, I want to put a record into either tableA or tableB.

Because s3 is used as an intermediary, I need to sort records into two files. How would I partition my buffer to do that?

[harlow]

@Diamonds0a I spiked out a prototype of splitting JSON events into several files on S3.

Here is the core of it:
https://github.com/harlow/event-stream/blob/master/src/com/eventstream/S3EventEmitter.java#L48-L97

The README has a few more details on how the JSON payloads should be structured:
https://github.com/harlow/event-stream

I haven't written the Redshift Load part of it yet.