Samza 流

所述samza容器读取和写入利用消息 SystemConsumer 和 SystemProducer 接口。您可以通过实现这两个接口与 Samza 集成任何消息代理。

public interface SystemConsumer {
  void start();

  void stop();

  void register(
      SystemStreamPartition systemStreamPartition,
      String lastReadOffset);

  List poll(
      Map systemStreamPartitions,
      long timeout)
    throws InterruptedException;
}

public class IncomingMessageEnvelope {
  public Object getMessage() { ... }

  public Object getKey() { ... }

  public SystemStreamPartition getSystemStreamPartition() { ... }
}

public interface SystemProducer {
  void start();

  void stop();

  void register(String source);

  void send(String source, OutgoingMessageEnvelope envelope);

  void flush(String source);
}

public class OutgoingMessageEnvelope {
  ...
  public Object getKey() { ... }

  public Object getMessage() { ... }
}

开箱即用，Samza 支持 Kafka（KafkaSystemConsumer 和 KafkaSystemProducer）。然而，可以插入任何消息总线系统，只要它可以提供 Samza 要求的语义，如javadoc中所述。

SystemConsumers 和 SystemProducer 可以读取和写入任何数据类型的消息。只要它们只支持字节数组就可以了 - Samza 有一个独立的序列化层，它转换为应用程序代码可以使用的对象。Samza 没有规定任何特定的数据模型或序列化格式。

作业配置文件可以包括特定于特定消费者和生产者实现的属性。例如，配置通常会指示要使用的消息代理的主机名和端口，也可能指示连接选项。

流如何处理

如果作业正在消耗来自多个输入流的消息，并且所有输入流都具有可用的消息，则默认情况下以循环方式处理消息。例如，如果作业正在消耗 AdImpressionEvent 和 AdClickEvent，则使用来自 AdImpressionEvent 的消息调用任务实例的 process（）方法，然后使用来自 AdClickEvent 的消息，然后来自 AdImpressionEvent 的另一个消息，并继续在两者之间交替。

如果其中一个输入流没有可用的新消息（最近的消息已经被消耗），则该流被跳过，并且该作业从其他输入继续消耗。它继续检查新消息是否可用。

MessageChooser

当 Samza 容器在不同流分区上有几个传入消息时，它如何决定首先处理哪个？行为由 MessageChooser 确定。默认选择器是 RoundRobinChooser，但您可以通过实现自定义选择器来覆盖它。

要插入自己的邮件选择器，您需要实现 MessageChooserFactory 界面，并将 “task.chooser.class” 配置设置为实施的完全限定类名称：

task.chooser.class=com.example.samza.YourMessageChooserFactory

优先输入流

某些时间来自一个流的消息应该比来自另一个流的消息优先处理。例如，一些 Samza 作业消耗两个流：一个流由实时系统馈送，另一个流由批处理系统馈送。在这种情况下，通过批处理流优先处理实时流是有用的，以便在批量流中突然发生数据突发时，实时处理速度不会降低。

Samza 提供了一种通过设置此配置参数来将一个流优先于另一个流的机制：systems. .streams. .samza.priority =

例如：

systems.kafka.streams.my-real-time-stream.samza.priority=2
systems.kafka.streams.my-batch-stream.samza.priority=1

这意味着，我的实时流的消息应该比我的批处理流的消息优先级更高。如果我的实时流有任何消息可用，它们将被先处理。只有当目前没有消息在等待我的实时流时，Samza 的工作才会继续处理我的批量流。

每个优先级别都有自己的 MessageChooser。定义具有相同优先级的两个流是有效的。如果消息从两个流以相同的优先级可用，则该优先级的 MessageChooser 将决定首先处理哪个消息。

仅定义某些流的优先级也是有效的。所有非优先流都被视为最低优先级，并共享一个 MessageChooser。

引导

有时，在处理来自任何其他流的消息之前，Samza 作业需要完全消耗流（从偏移0到最近的消息）。在流包含作业需要的一些必备数据的情况下，这是非常有用的，在作业加载了先决条件数据之前处理来自其他流的消息是没有意义的。Samza 支持这种使用引导流的用例。

引导流似乎与具有高优先级的流相似，但是略有不同。在允许任何其他流被处理之前，引导流等待消费者明确地确认流已被完全消耗。在此之前，引导流是作业的独占输入：即使网络问题或其他因素导致引导流消费者放慢速度，其他输入也不能潜入其中。

引导流和高优先级流之间的另一个区别是引导流的特殊处理是临时的：当它被完全消耗（我们说它已经被“捕获”）时，其优先级与所有其他流相同输入流。

要将名为“my-bootstrap-stream”的流配置为完全消耗的引导流，请使用以下设置：

systems.kafka.streams.my-bootstrap-stream.samza.bootstrap=true
systems.kafka.streams.my-bootstrap-stream.samza.reset.offset=true
systems.kafka.streams.my-bootstrap-stream.samza.offset.default=oldest

bootstrap = true 参数启用引导行为（优先于其他流）。reset.offset = true和offset.default = oldest 的组合告诉 Samza 始终从最早的偏移量开始读取流，每次容器启动（而不是从最近的检查点开始读取）。

定义多个引导流是有效的。在这种情况下，它们被引导的顺序由优先级确定。

配料

在某些情况下，您可以按顺序从相同的流分区中消耗多个消息来提高性能。Samza 支持这种操作模式，称为批处理。

例如，如果要从每个流分区（不管 MessageChooser）读取一行中的 100 条消息，可以使用此配置参数：

task.consumer.batch.size=100

使用此设置，Samza 尝试从最近使用的 SystemStreamPartition中 读取消息。直到没有更多消息可用于该 SystemStreamPartition，或者直到达到批量大小，此行为将继续。当这种情况发生时，Samza 将前往 MessageChooser 来确定要处理的下一个消息。然后再次尝试从所选消息的 SystemStreamPartition 继续消费，直到达到批量大小。

关注微信下载离线手册