Samza 3---Samza API

Samza任务类必须实现StreamTask,可选 InitableTask, ClosableTask,WindowableTask

public class xxxxTask implements StreamTask, InitableTask, ClosableTask,WindowableTask {

}

StreamTask

public interface StreamTask {
    void process(IncomingMessageEnvelope var1, //接收到的消息封装
                MessageCollector var2,//用来发送其他消息
                TaskCoordinator var3) 
    throws Exception;
}

IncomingMessageEnvelop

代表接收到的消息的封装,表示StreamTask收到一个分区的一个特定的输入流。包括：

• message
• key
• 消息来源（system+stream+partition）

  public class IncomingMessageEnvelope {
    /** A deserialized message. */
    Object getMessage() { ... }
  
    /** A deserialized key. */
    Object getKey() { ... }
  
    /** The stream and partition that this message came from. */
    SystemStreamPartition getSystemStreamPartition() { ... }
  }

SystemStreamPartition

消息来源包括：

• 消息来源system名
• 消息来源stream/topic/queue名
• 消息流的分区

  /** A triple of system name, stream name and partition. */
  public class SystemStreamPartition extends SystemStream {
  
    /** The name of the system which provides this stream. It is
        defined in the Samza job's configuration. */
    public String getSystem() { ... }
  
    /** The name of the stream/topic/queue within the system. */
    public String getStream() { ... }
  
    /** The partition within the stream. */
    public Partition getPartition() { ... }
    
  }

MessageCollector

为了发送一个消息， 你会创建一个OutgoingMessageEnvelop对象并且把它传递给消息收集器。它至少会确定你想要发送的消息、系统和数据流名字再发送出去。你也可以确定分区的key和另一些参数。

public interface MessageCollector {
  void send(OutgoingMessageEnvelope envelope);
}

注意：
请只在process()方法里使用MessageCollector对象。如果你保持住一个MessageCollector实例并且之后再次使用它，你的消息可能会错误地发送出去。

OutgoingMessageEnvelope

public class OutgoingMessageEnvelope {
    private final SystemStream systemStream;
    private final String keySerializerName;
    private final String messageSerializerName;
    private final Object partitionKey;
    private final Object key;
    private final Object message;
    
    public OutgoingMessageEnvelope(SystemStream systemStream, String keySerializerName, String messageSerializerName, Object partitionKey, Object key, Object message) { ... }
    
     public OutgoingMessageEnvelope(SystemStream systemStream, Object partitionKey, Object key, Object message) { ... }

    public OutgoingMessageEnvelope(SystemStream systemStream, Object key, Object message) { ... }
    
    public OutgoingMessageEnvelope(SystemStream systemStream, Object message) { ... }
}

InitableTask

用来处理初始化工作，init 函数会首先被调用

public interface InitableTask {
    void init(Config var1, TaskContext var2) throws Exception;
}

ClosableTask

public interface ClosableTask {
    void close() throws Exception;
}

WindowableTask

public interface WindowableTask {
    void window(MessageCollector var1, 
                TaskCoordinator var2) throws Exception;
}

实例

有一个简单的任务，它把每个输入的消息拆成单词，并且发送每一个单词作为一个消息

 public class SplitStringIntoWords implements StreamTask, InitableTask, ClosableTask {

    private String outputSystem;
    
    //官网教程中直接使用固定的SystemStream,稍加改造
    //private final SystemStream output_stream = new SystemStream("kafka", "words");
    
    @Override
    public void close() throws Exception {
    }
    
    @Override
    public void init(Config config, TaskContext context) throws Exception {//初始化时从配置文件读output system
        outputSystem = config.get("task.output.system");
    }
    
    public void process(IncomingMessageEnvelope envelope,
                      MessageCollector collector,
                      TaskCoordinator coordinator) {
        String topic = envelope.getSystemStreamPartition().getSystemStream().getStream();
        
        topic=topic+"_split";//获取输入消息的topic，加上后缀
        
        SystemStream output_stream=new SystemStream(outputSystem, topic);//构造新的发送消息的System+Stream
        
        String message = (String) envelope.getMessage();//接收到的消息
        
        for (String word : message.split(" ")) {
          //单词作为key，1作为value，后续任务可以将1相加计数
          collector.send(new OutgoingMessageEnvelope(output_stream, word, 1));
        }
    }
}

参考：

https://samza.apache.org/learn/documentation/0.10/
https://github.com/feuyeux/hello-samza/blob/master/doc/0.hello_samza.md
http://wdxtub.com/vault/cc-21.html
http://www.infoq.com/cn/articles/linkedin-samza
http://blog.csdn.net/colorant/article/details/12082145
https://wyyhzc.gitbooks.io/hadoop2x/content/samzacontainer.html