Samza附录十一 Samza配置参考
下表列出了可以包含在Samza作业配置文件中的所有标准属性。
下表中加粗显示的 system-name 是您自己的变量名的占位符。
| 名称 | 默认 | 描述 |
|---|---|---|
| Samza工作配置 | ||
| job.factory.class |
必需：用于运行此作业的作业工厂。该值是一个完全限定的Java类名，必须实现 StreamJobFactory。Samza 有三个实现：
|
|
| job.name | 必需：您的工作的名称。该名称出现在 Samza 仪表板上，用于从其他工作的检查点分离此工作的检查点。 | |
| job.id | 1 |
如果你同时运行几个你的工作实例，你需要给每个执行一个不同的job.id。这很重要，否则这些工作将覆盖彼此的检查点，也可能以其他方式相互干扰。
|
| job.coordinator.system | 必需：用于创建和维护协调器流的 system-name。 | |
| job.default.system | 该 system-name 访问系统没有明确配置用于其中的任何输入或输出流。此属性用于输入和输出流，而job.coordinator.system 适用于 samza 元数据流。 | |
job.coordinator.replication.factor
|
3 | 如果您使用 Kafka 进行协调器流，那么您希望协调器主题复制其持久性的 Kafka 节点的数量。 |
job.coordinator.segment.bytes
|
26214400 | 如果您使用 Kafka 系统进行协调器流，则这是用于协调器主题日志段的段大小。保持这个数字很小是有用的，因为它增加了Kafka 垃圾收集旧消息的频率。 |
job.coordinator.monitor-partition-change
|
假 | 如果您使用 Kafka 进行协调器流，则此配置可使作业协调器检测 Kafka 输入主题中的分区计数差异。检测时，它会更新system-name 格式的 Gauge 指标。stream-name .partitionCount，表示分区计数与初始状态的差异。请注意，目前此功能仅适用于基于卡夫卡的系统。 |
job.coordinator.monitor-partition-change.frequency.ms
|
300000 | 应检测输入流分区计数变化的频率。由于分区增加不是常见事件，所以可以将此检查调整为相当低的。 |
job.config.rewriter.rewriter-name.class
|
您可以选择定义配置重写器，有可能在作业启动前动态修改作业配置。例如，这可以用于从外部配置管理系统中拉取配置，或者用于在运行时动态地确定输入流集合。该属性的值是一个完全限定的Java类名，它必须实现 ConfigRewriter。默认情况下，Samza与这些重写器一起发货：
|
|
| job.config.rewriters | 如果您已定义配置重写器，则需要按照应用顺序将它们列在这里。此属性的值为以逗号分隔的重写器名称令牌列表 。 | |
job.systemstreampartition.grouper.factory
|
org.apache.samza.container.grouper.stream.GroupByPartitionFactory
|
用于确定如何将SystemStreamPartition输入组合在一起以在单个StreamTask实例中处理的工厂类。工厂必须实现SystemStreamPartitionGrouperFactory接口。一旦设置了这个配置，就不能改变，因为这样做可能会违反状态语义，并导致数据丢失。
|
job.systemstreampartition.matcher.class
|
如果要启用静态分区分配，则这是必需的配置。此属性的值是实现该接口的完全限定的 Java 类名称 org.apache.samza.system.SystemStreamPartitionMatcherSamza 配有两个匹配课：
|
|
job.systemstreampartition.matcher.config.range
|
如果job.systemstreampartition.matcher.class指定，并且该属性的值是 org.apache.samza.system.RangeSystemStreamPartitionMatcher，则此属性是 必需的配置。指定一个逗号分隔的范围列表，以确定哪个分区匹配，从而静态分配给作业。例如“2,3,11-20”，对于所有指定的系统和流（在Kafka的情况下的主题），为该作业静态分配2,3和11到20。像“19”这样的 singel 配置值也是有效的。这个静态分配分区19。对于 config 验证，逗号分隔列表中的每个元素都很符合以下正则表达式之一：
|
|
job.systemstreampartition.matcher.config.regex
|
如果job.systemstreampartition.matcher.class指定，并且该属性的值是 org.apache.samza.system.RegexSystemStreamPartitionMatcher，则此属性是 必需的配置。该值应该是一个有效的 Java 支持的正则表达式。例如 “[1-2]”，将所有指定的系统和流（Kakfa 的情况下的主题）的分区1和2静态分配给作业。
|
|
job.systemstreampartition.matcher.config.job.factory.regex
|
此配置可用于指定 Java 支持的正则表达式，以匹配StreamJobFactory 应启用静态分区分配的正则表达式。此配置使分区分配功能也可用于自定义StreamJobFactory。此配置默认为以下值： |
|
job.checkpoint.validation.enabled
|
真 |
此设置控制作业是否应该失败（true）或只是警告（false），以防止检查点分区号验证失败。 注意：此配置需要谨慎使用。在检查点自动创建错误数量的分区之后，它应该仅用作解决方法。 |
| job.security.manager.factory | 这是用于创建适当的 SecurityManager 的 factory 类，用于在安全环境中运行时处理 Samza 容器的安全性，例如使用Kerberos 进行的纱线。默认情况下： | |
| job.container.count | 1 | 要求运行您的工作的YARN容器的数量。这是控制作业规模（分配计算资源）的主要参数：为了增加处理的并行性，您需要增加容器数量。最小值是一个容器，容器的最大数量是任务实例的数量（通常是输入流分区的 数量）。任务实例均匀分布在您指定的容器数量上。 |
| job.container.single.thread.mode | 假 | 如果设置为 true，samza 将回退到传统的单线程事件循环。默认值为 false，这将启用多线程执行。 |
| job.container.thread.pool.size | 如果已配置，容器线程池将用于并行运行每个任务的同步操作。操作包括 StreamTask.process（），WindowableTask.window（）和内部的 Task.commit（）。请注意，线程池不适用于AsyncStremTask.processAsync（）。大小应该始终大于零。如果未配置，则所有任务操作都将在单个线程中运行。 | |
| job.host-affinity.enabled | 假 | 此属性指示是否启用主机关联。主机关联是指 Samza 每次部署作业时在同一台主机上请求和分配容器的能力。当启用主机密切关系时，Samza 做出了“尽力而为”来遵守主机关联约束。属性 cluster-manager.container.request.timeout.ms 确定在对主机相关性约束进行取消优先级排序并将容器分配给任何可用资源之前等待多长时间。 请注意：启用连续调度后，该功能将经过 yarn 中的 FairScheduler 测试。 |
| job.changelog.system | 此属性指定一个用于 changelog 的默认系统，它将与 stores.store-name.changelog config中指定的流一起使用 。您可以通过在 stores.store-name.changelog 中指定系统和流来覆盖此系统 。 | |
| job.coordinator.factory |
用于工作协调的类。目前可用的值为：
|
|
| 基于Zookeeper的作业配置 | ||
| job.coordinator.zk.connect | 对于使用基于 Zookeeper 协调的应用程序而言是必需的。Zookeeper 坐标（在“host：port [/ znode]”格式中）用于协调。 | |
| job.coordinator.zk.session.timeout.ms | 30000 | 所有 ZK 连接的 Zookeeper 会话超时以毫秒为单位。会话超时控制在不能与 ZK 服务器通话之前，zk 客户端在抛出异常之前等待多久。 |
| job.coordinator.zk.connection.timeout.ms | 60000 | Zookeeper 连接超时（毫秒）。Zk 连接超时控制客户端在放弃之前尝试连接到 ZK 服务器的时间。 |
| job.coordinator.zk.consensus.timeout.ms | 40000 | 每个处理器将等待所有处理器在回滚之前报告接受新作业模型的时间。 |
| job.debounce.time.ms | 2000 | 在注册处理器更改之前，Leader 处理器需要等待多久才能重新计算 JobModel。 |
| 任务配置 | ||
| task.class |
必需：从输入流处理传入消息的 Java 类的完全限定名称。这个类必须实现 StreamTask 或 AsyncStreamTask，并且可任选地实现 InitableTask， ClosableTask 和 WindowableTask。该类将被实例化几次，每次输入流分区一次 。 |
|
| task.inputs |
必需：以此作业使用的流的逗号分隔列表。每个流都以system-name 格式给出 stream-name 。举例来说，如果你有一个叫做输入系统my-kafka，并要消耗2个 Kafka 主题叫做PageViewEvent和UserActivityEvent，那么你就设置 task.inputs=my-kafka.PageViewEvent, my-kafka.UserActivityEvent。
|
|
| task.window.ms | -1 | 如果 task.class 实现 WindowableTask，它可以定期接收一个 加窗回调。此属性指定window（）调用之间的时间，以毫秒为单位。如果数字为负数（默认值），则不会调用window（）。请注意，Samza是 单线程的，所以window（）调用永远不会与处理消息同时发生。如果在window（）调用到期时正在处理消息，则在完成当前消息的处理之后发生window（）调用。 |
| task.checkpoint.factory |
要启用 检查点，必须将此属性设置为实现CheckpointManagerFactory 的 Java 类的完全限定名称 。这不是必需的，但建议大多数工作。如果您没有配置检查点，并且作业或容器重新启动，它不记得它已经处理了哪些消息。没有检查点，消费者行为由 ... samza.offset.default 决定设置，默认情况下将跳过在容器重新启动时发布的任何消息。检查点允许作业启动它以前停止的位置。默认情况下，Samza 有两个检查点管理员：
|
|
| task.commit.ms | 60000 | 如果配置了 task.checkpoint.factory，则此属性确定检查点的写入频率。值是检查点之间的时间，以毫秒为单位。检查点的频率会影响故障恢复：如果容器意外失败（例如，由于崩溃或机器故障）而重新启动，则会在最后一个检查点恢复处理。从失败的容器上的最后一个检查点处理的任何消息将被再次处理。检查点更频繁地减少可能被处理两次的消息数量，而且还使用更多的资源。 |
| task.command.class |
org.apache.samza.job.ShellCommandBuilder
|
确定 容器 的命令行和环境变量的Java类的完全限定名称。它必须是 CommandBuilder 的子类 。默认为task.command.class=org.apache.samza.job.ShellCommandBuilder。
|
| task.opts |
任何 JVM 选项在执行Samza容器时都包含在命令行中。例如，这可以用于设置JVM堆大小，调整垃圾回收器或启用 远程调试。运行时无法使用ThreadJobFactory。任何您放入的东西 task.opts都将直接转发到命令行，作为JVM调用的一部分。
|
|
| task.java.home |
Samza 容器的 JAVA_HOME 路径。通过设置此属性，可以使用与集群的 Java 版本不同的 java 版本。记住设置yarn.am.java.home好。
|
|
| task.execute | bin/ run-container.sh | 启动 Samza 容器的命令。脚本必须包含在 作业包中。通常不需要定制这个。 |
| task.chooser.class |
org.apache.samza.system.chooser.RoundRobinChooserFactory
|
该属性可以选择设置为覆盖默认 消息选择器，该消息选择器确定处理来自多个输入流的消息的顺序。此属性的值是实现MessageChooserFactory 的 Java 类的完全限定名称 。 |
| task.drop.deserialization.errors | 这个属性是定义系统如何处理反序列化失败的情况。如果设置为 true，系统将跳过错误消息并继续运行。如果设置为false，则系统将抛出异常并使容器失败。默认值为 false。 | |
| task.drop.serialization.errors | 此属性用于定义系统如何处理序列化失败情况。如果设置为true，系统将丢弃错误消息并继续运行。如果设置为 false，则系统将抛出异常并使容器失败。默认值为 false。 | |
| task.log4j.system |
指定 StreamAppender 的系统名称。如果配置中未指定此属性，则 Samza 将抛出异常。（参见 Stream Log4j Appender）
|
|
| task.log4j.location.info.enabled | 假 |
定义是否在 Log4j StreamAppender 消息中包含 log4j 的LocationInfo 数据。LocationInfo 包括写入日志消息的文件，类和行等信息。此设置仅在使用Log4j流附加程序时有效。（参见Stream Log4j Appender）
|
| task.poll.interval.ms | Samza 的容器在两个条件下轮询更多的消息。当任何输入SystemStreamPartition 不存在任何剩余的缓冲消息时，第一个条件出现。第二个条件出现当一些输入SystemStreamPartitions 有空缓冲区，但有些不具有。在后一种情况下，定义轮询间隔以确定刷新空的SystemStreamPartition 缓冲区的频率。默认情况下，此间隔为 50ms，这意味着任何空的 SystemStreamPartition 缓冲区至少每 50ms 刷新一次。这里的值越大意味着空的SystemStreamPartition 将不会更频繁地刷新，这意味着更多的延迟被引入，但是将使用更少的 CPU 和网络。 | |
| task.ignored.exceptions |
此属性指定在任务process或window 方法中抛出哪些异常应忽略。被忽略的例外应该是一个逗号分隔的例外的完全限定类名的列表，或 *忽略所有异常。
|
|
| task.shutdown.ms | 5000 | 此属性控制 Samza 容器等待有序关闭任务实例的时间。 |
| task.name.grouper.factory |
org.apache.samza.container.grouper.task.GroupByContainerCountFactory
|
确定将生成 TaskNameGrouper 的工厂类的 Java 类的完全限定名称。如果属性不存在，则默认配置值为task.name.grouper.factory=org.apache.samza.container.grouper.task.GroupByContainerCountFactory用户可以指定 TaskNameGrouperFactory 的自定义实现，其中实现用于对任务进行分组的自定义逻辑。 注意：对于非集群应用程序（使用协调服务的应用程序），必须使用 org.apache.samza.container.grouper.task.GroupByContainerIdsFactory |
| task.broadcast.inputs |
此属性指定所有任务应消耗的分区。你把这里的systemStreamPartitions 发送到所有的任务。
|
|
| task.max.concurrency | 1 |
每个任务一次处理的最大数量的未完成的消息，它适用于StreamTask 和 AsyncStreamTask。值可以是：
|
| task.callback.timeout.ms | 此属性仅适用于 AsyncStreamTask。它定义从processAsync（）到回调的最大时间间隔被触发。当超时发生时，它将抛出一个 TaskCallbackTimeoutException 并关闭该容器。默认是没有超时。 | |
| task.consumer.batch.size | 1 | 如果设置为正整数，则任务将尝试使用 每个输入流中给定数量的消息的批次，而不是从每个单独消息的所有输入流中消耗循环。在某些情况下，设置此属性可以提高性能。 |
| 系统 | ||
systems.system-name.samza.factory
|
必需：提供系统的 Java 类的全限定名称 。系统可以提供您可以在 Samza 作业中使用的输入流，也可以输出可以写入的输出流，或同时提供两者。对系统的要求非常灵活 - 它可以连接到消息代理，或读取和写入文件，或使用数据库或其他任何东西。该类必须实现 SystemFactory。Samza 具有以下实现：
|
|
systems.system-name.default.stream*
|
与系统关联的任何流的一组默认属性。例如，如果配置了“systems.kafka-system.default.stream.replication.factor”= 2，则在kafka 系统上创建的每个 Kafka 流将具有2的复制因子，除非在流中显式覆盖该属性范围使用流属性。 | |
|
systems.system-name.
default.stream.samza.key.serde
|
该 SERDE 将用于反序列化 键上输入流的消息，和序列化键上输出流消息。此属性定义系统中所有流的serde。请参阅 stream-scoped 属性来定义单个流的 serde。如果两者都被定义，流级定义优先。此属性的值必须是使用 serializers.registry 注册的 serde-namesystems.system-name.*。如果未设置此属性，消息将在输入流消费者，任务和输出流生成器之间未修改传递。 | |
|
systems.system-name.
default.stream.samza.msg.serde
|
将用于反序列化输入流上的消息值 的 serde，并串行化输出流上消息的值。此属性定义系统中所有流的serde。请参阅 stream-scoped 属性来定义单个流的 serde。如果两者都被定义，流级定义优先。此属性的值必须是使用serializers.registry 注册的 serde-name *。如果未设置此属性，消息将在输入流消费者，任务和输出流生成器之间未修改传递。 | |
|
systems.system-name.
default.stream.samza.offset.default
|
即将到来 |
如果容器在没有检查点的情况下启动，则该属性确定输入流中我们应该开始使用的位置。该值必须为 OffsetType，为以下之一：
samza.offset.default 如果两者都被定义，流级定义优先。 |
systems.system-name.samza.key.serde
|
这不支持 systems.system-name.default.stream.samza.key.serde | |
|
systems.system-name.streams.stream-name.
samza.key.serde
|
这不支持 streams.stream-id.samza.key.serde. | |
systems.system-name.samza.msg.serde
|
这不支持 systems.system-name.default.stream.samza.msg.serde | |
|
systems.system-name.streams.stream-name.
samza.msg.serde
|
这不支持 streams.stream-id.samza.msg.serde | |
systems.system-name.samza.offset.default
|
即将到来 | 这不支持 systems.system-name.default.stream.samza.offset.default. |
|
systems.system-name.streams.stream-name
samza.offset.default
|
这不支持 streams.stream-id.samza.offset.default | |
|
systems.system-name.streams.stream-name
samza.reset.offset
|
假 | 这不支持 streams.stream-id.samza.reset.offset. |
|
systems.system-name.streams.stream-name
samza.priority
|
-1 | 这不支持 streams.stream-id.samza.priority. |
|
systems.system-name.streams.stream-name
samza.bootstrap
|
假 | 这不支持 streams.stream-id.samza.bootstrap. |
| 流 | ||
streams.stream-id.samza.system
|
将访问此流的系统的 system-name。此属性将流绑定到由属性系统定义的系统之一。system-name.samza.factory。 如果未指定此属性，它将从 job.default.system 继承。 |
|
streams.stream-id.samza.physical.name
|
系统上将访问此流的物理名称。这是与 Samza 用来识别流的逻辑名称的 stream-id 相反的。物理名称可以是 Kafka 主题名称，HDFS 文件 URN 或任何其他系统特定的标识符。 | |
streams.stream-id.samza.key.serde
|
该 SERDE 将用于反序列化 键上输入流的消息，和序列化键上输出流消息。此属性定义单个流的serde。请参阅 system-scoped 属性以定义系统中所有流的 serde。如果两者都被定义，流级定义优先。此属性的值必须是使用 serializers.registry 注册的 serde-name.*。如果未设置此属性，消息将在输入流消费者，任务和输出流生成器之间未修改传递。 | |
streams.stream-id.samza.msg.serde
|
将用于反 序列化输入流上的消息值的 serde，并串行化输出流上消息的值。此属性定义单个流的serde。请参阅 system-scoped 属性以定义系统中所有流的serde。如果两者都被定义，流级定义优先。此属性的值必须是使用serializers.registry 注册的 serde-name .*。如果未设置此属性，消息将在输入流消费者，任务和输出流生成器之间未修改传递。 | |
streams.stream-id.samza.offset.default
|
即将到来 |
如果容器在没有检查点的情况下启动，则该属性确定输入流中我们应该开始使用的位置。该值必须为 OffsetType，为以下之一：
|
streams.stream-id.samza.reset.offset
|
假 |
如果设置为true，当Samza容器启动时，它将忽略该特定输入流的任何 检查点偏移量。因此，其行为由samza.offset.default设置决定。请注意，每次启动容器时，复位都将生效，这可能是每次重新启动作业时，或者如果容器发生故障并由框架重新启动，则会更频繁。
|
streams.stream-id.samza.priority
|
-1 | 如果一个或多个流具有优先级设置（任何正整数），则将以比其他流更高的优先级处理它们。您可以将多个流设置为相同的优先级，或通过为较高优先级的流分配更多的数字来定义多个优先级。如果较高优先级的流具有可用的消息，则它们将始终被处理; 来自较低优先级流的消息仅在高优先级输入没有新消息时被处理。 |
streams.stream-id.samza.bootstrap
|
假 |
如果设置为true，该流将被作为streams.自举流处理。这意味着每次 Samza 容器启动时，在处理任何其他流的消息之前，该流将被完全消耗。
|
| streams.stream-id.* | 流的任何属性。这些通常是系统特定的，可以由系统用于流创建或验证。请注意，其他属性前缀为 samza。 将它们区分为不是系统特定的 Samza 属性。 | |
| 串行器/解串器（Serdes） | ||
serializers.registry.serde-name .class
|
使用此属性注册序列化器/解串器，它定义了将应用对象编码为字节数组（用于流中的消息以及持久存储中的数据）的方式。你可以给你所需要的任何 serde-serde 名字，并在系统的属性中引用该名称 systems.*.samza.key.serde， streams.*.samza.key.serde，streams.*.samza.msg.serde， stores.*.key.serde systems.*.samza.msg.serde，streams.*.samza.key.serde， streams.*.samza.key.serde， streams.*.samza.msg.serde，stores.*.key.serde 和 stores.*.msg.serde。此属性的值是实现 SerdeFactory 的 Java 类的完全限定名称 。Samza 有几个 serdes：
|
|
|
将文件系统用于检查点
（本节适用于您已设置 task.checkpoint.factory
= org.apache.samza.checkpoint.file.FileSystemCheckpointManagerFactory） | ||
| task.checkpoint.path | 如果将文件系统用于检查点，则为必需。将其设置为本地文件系统上应存储检查点文件的路径。 | |
|
使用 Elasticsearch 输出流
（适用于您已设置 systems.*.samza.factory
= org.apache.samza.system.elasticsearch.ElasticsearchSystemFactory） | ||
systems.system-name.client.factory
|
必需：弹性搜索客户端工厂用于连接到弹性搜索集群。Samza具有以下实现：
|
|
systems.system-name.index.request.factory
|
org.apache.samza.system.elasticsearch.indexrequest.
DefaultIndexRequestFactory
|
索引请求工厂将Samza OutgoingMessageEnvelope转换为IndexRequest以发送到弹性搜索。默认的IndexRequestFactory的行为如下：
|
systems.system-name.client.transport.host
|
需要的TransportClientFactoryTransportClientFactory 连接到的主机名。 |
|
systems.system-name.client.transport.port
|
需要的TransportClientFactoryTransportClientFactory 连接到的端口。 |
|
systems.system-name.client.elasticsearch.*
|
任何 Elasticsearch 客户端设置都可以在这里使用。它们都将传递给传输和节点客户端。您想要提供的一些常见设置是。
|
|
systems.system-name.bulk.flush.max.actions
|
1000 | 冲洗前要缓冲的最大消息数。 |
systems.system-name.bulk.flush.max.size.mb
|
五 | 冲洗前缓冲的消息的最大聚合大小。 |
systems.system-name.bulk.flush.interval.ms
|
决不 | 缓冲消息应该被刷新多久。 |
|
使用 Kafka 输入流，输出流和检查点
（适用于您已设置 systems.*.samza.factory
= org.apache.samza.system.kafka.KafkaSystemFactory） | ||
| systems.system-name.consumer.zookeeper.connect |
可以找到有关 Kafka 群集信息的一个或多个 Zookeeper 节点的主机名和端口。这是以逗号分隔的hostname:port对列表给出的 ，例如zk1.example.com:2181,zk2.example.com:2181,zk3.example.com:2181。如果集群信息位于 Zookeeper 命名空间的某个子路径，则需要将路径包含在主机名列表的末尾，例如：zk1.example.com:2181,zk2.example.com:2181,zk3.example.com:2181/clusters/my-kafka
|
|
| systems.system-name.consumer.auto.offset.reset | 最大 |
此设置确定使用者如何尝试读取超出当前有效范围的偏移量会发生什么。如果主题不存在，或者如果检查点比代理所保留的最大消息历史更早，则可能会发生这种情况。这个属性不要与 systems.*.samza.offset.default 混淆，它确定如果没有检查点会发生什么。以下是以下的有效值auto.offset.reset：
|
| systems.system-name.consumer.* |
任何 Kafka 使用者配置 都可以包含在这里。例如，要更改套接字超时，可以设置系统。systemname .consumer.socket.timeout.ms。（没有必要配置，group.id或者client.id是由Samza自动配置，而且没有必要设置， auto.commit.enable因为Samza 有自己的检查点机制）
|
|
| systems.system-name.producer.bootstrap.servers |
注意： 此变量之前已定义为“producer.metadata.broker.list”，该版本已被弃用。 Kafka 代理正在运行的网络端点列表。例如，以逗号分隔的 hostname:port对列表给出 kafka1.example.com:9092,kafka2.example.com:9092,kafka3.example.com:9092。不需要列出集群中的每个单个Kafka节点：Samza使用此属性来发现哪些主题和分区托管在哪个代理上。即使您只是从卡夫卡消费而不是写信给这个属性，因为 Samza 使用它来发现关于正在消费的流的元数据。
|
|
|
systems.system-name.producer.producer.*
|
这里可以包含 任何 Kafka 生产者配置。例如，要更改请求超时，可以设置系统。system-name .producer.timeout.ms。（没有必要配置，client.id因为它是由Samza自动配置的。）
|
|
| systems.system-name.samza.fetch.threshold | 50000 | 当从 Kafka 消耗流时，Samza 容器为传入的消息维护一个内存中的缓冲区，以增加吞吐量（流任务可以继续处理缓冲的消息，而从 Kafka 获取新消息）。此参数确定了我们旨在缓冲容器所消耗的所有流分区的消息数。例如，如果一个容器消耗50个分区，它将默认缓存每个分区的1000个消息。当缓冲消息的数量低于该阈值时，Samza 从 Kafka 代理获取更多消息来补充缓冲区。增加此参数可以增加作业“ |
| systems.system-name.samza.fetch.threshold.bytes | -1 |
当从 Kafka 消耗流时，Samza 容器为传入的消息维护一个内存中的缓冲区，以增加吞吐量（流任务可以继续处理缓冲的消息，而从 Kafka 获取新消息）。此参数确定了我们旨在基于字节在容器所消耗的所有流分区之间缓冲的消息的总大小。定义作为整体的缓冲预取消息使用的字节数。基于此计算单个系统/流/分区的字节。这将获取整个消息，因此这个字节限制是一个软的，并且实际使用可以是给定流的分区中的最大消息的字节数限制+大小。如果此属性的值为大于 0，那么这将优先于系统。system-name.samza.fetch.threshold。 例如, 如果 fetchThresholdBytes 设置为100000个字节, 并且注册了 50 SystemStreamPartitions, 则每个分区的阈值为 (100000/2)/50 = 1000 字节。由于这是一个软限制, 实际使用可以是1000字节 + 最大消息的大小。一旦 SystemStreamPartition 的缓冲消息字节数降至 1000 以下, 就会执行回迁请求以获取更多数据。增加此参数将减少当队列被消息耗尽和新消息排队时之间的延迟, 但也会导致内存使用量增加, 因为将有更多的消息保存在内存中。默认值为-1, 表示不使用此方法。 |
| task.checkpoint.system |
如果您使用Kafka检查点（task.checkpoint.factory = org.apache.samza.checkpoint.kafka.KafkaCheckpointManagerFactory），则此属性是必需的。您必须将其设置为 Kafka 系统的系统名称。该系统中的流名称（主题名称）将根据作业名称和ID自动确定：（ 在作业名称中带下划线，ID由连字符替换）。 __samza_checkpoint_${job.name}_${job.id}
|
|
task.checkpoint.replication.factor
|
3 | 如果您使用 Kafka 作为检查点，那么您希望将检查点主题复制到其持久性的 Kafka 节点的数量。 |
| task.checkpoint.segment.bytes | 26214400 | 如果您使用 Kafka 作为检查点，则这是用于检查点主题的日志段的段大小。保持这个数字很小是有用的，因为它增加了卡夫卡垃圾收集旧检查点的频率。 |
| stores.store-name .changelog.replication.factor | stores.default.changelog.replication.factor | 该属性定义要用于更改日志流的副本数。 |
| stores.default.changelog.replication.factor | 2 | 此属性定义要用于更改日志流的副本的默认数量。 |
| stores.store-name .changelog.kafka.topic-level-property | 该属性允许您为要创建的更改日志主题指定主题级别设置。例如，您可以将清理策略指定为“stores.mystore.changelog.cleanup.policy = delete”。有关更多主题级配置，请参阅http://kafka.apache.org/documentation.html#configuration。 | |
|
使用与正则表达式匹配的所有 Kafka 主题
（本节适用于已设置 job.config.rewriter.*.class
= org.apache.samza.config.RegExTopicGenerator） | ||
| job.config.rewriter.rewriter-name.system | 将此属性设置为要从其中消费所有匹配主题的 Kafka 系统的system-name。 | |
| job.config.rewriter.rewriter-name.regex | 一个正则表达式，指定要在 Kafka 系统中使用哪些主题 job.config.rewriter.*.system。除了使用 task.inputs 指定的任何主题之外，与此正则表达式匹配的任何主题都将被使用。 | |
| job.config.rewriter.rewriter-name.config.* |
在此命名空间中指定的任何属性将应用于与job.config.rewriter.*.regex中的正则表达式匹配的流的配置 。例如，您可以设置job.config.rewriter.*.config.samza.msg.serde为匹配流中的邮件配置解串器，这相当于为 正则表达式匹配的每个主题设置 sytems.*.streams.*.samza.msg.serde。
|
|
| 存储和国家管理 | ||
| stores.store-name.factory |
这个属性定义了一个商店，Samza的有效状态流处理机制 。你可以给一家商店的任何商店的名字，除了默认（在商店名称 默认保留定义默认存储参数），并使用该名称在您的流任务中获得的存储的引用（在你的任务的 init（）方法调用 TaskContext.getStore（））。此属性的值是实现的Java类的完全限定名称 StorageEngineFactory。Samza目前随附一个存储引擎实现：
|
|
| stores.store-name.key.serde | 如果存储引擎预期存储中的密钥是简单的字节数组，则该 serde 允许流任务使用另一对象类型作为密钥访问存储。此属性的值必须是使用 serializers.registry 注册的 serde-name.*。如果没有设置该属性，键传递未修改存储引擎（和 更新日志流，如果合适的话）。 | |
| stores.store-name.msg.serde | 如果存储引擎预期商店中的值是简单的字节数组，则该 serde允许流任务使用另一个对象类型作为值访问存储。此属性的值必须是使用 serializers.registry 注册的 serde-name.*。如果未设置此属性，则将值修改为存储引擎（如果适用），则更改日志流。 | |
| stores.store-name.changelog | Samza 商店是集装箱本地的。如果容器发生故障，则商店的内容将丢失。为了防止数据丢失，您需要将此属性设置为配置更改日志流：Samza 然后确保对存储的写入将复制到此流，并在故障后从该流恢复存储。该属性的值以 system-name.stream-name的形式给出。“系统名称”部分是可选的。如果省略，则必须在 job.changelog.system 中指定系统配置。任何输出流都可以用作更改日志，但是您必须确保只有一个作业可以写入给定的更改日志流（作业的每个实例，每个存储需要自己的更改日志流）。 | |
|
使用 RocksDB 进行键值存储
（本节适用于您设置了 stores.*.factory
= org.apache.samza.storage.kv.RocksDbKeyValueStorageEngineFactory） | ||
| stores.store-name.write.batch.size | 500 | 为了获得更好的写入性能，存储引擎将缓冲区写入并将其应用到底层存储中。如果快速连续写入相同的密钥多次，则该缓冲区也会对相同的密钥进行重复数据删除。此属性设置为每个任务实例中应保存在此内存缓冲区中的键/值对数。数量不能大于 stores.*.object.cache.size |
| stores.store-name.object.cache.size | 1000 | Samza在经常访问的对象之前为RocksDB提供了额外的缓存。与缓存序列化对象的RocksDB块缓存（stores.*.container.cache.size.bytes）相反，此缓存包含反序列化对象（避免缓存命中中的反序列化开销）。此属性确定每个任务实例在Samza缓存中保留的对象数。同样的缓存也用于写入缓冲（请参阅 stores.*.write.batch.size）。值为 0 将禁用所有缓存和批处理。 |
| stores.store-name.container.cache.size.bytes | 104857600 | RocksDB的块缓存大小（以字节为单位），每个容器。如果一个容器内有若干个任务实例，则每个容器中都有一个比例份额。请注意，这是一个非堆内存分配，因此容器的总内存使用量是最大的JVM堆大小加上此缓存的大小。 |
| stores.store-name.container.write.buffer.size.bytes | 33554432 | 在每个容器写入磁盘之前，RocksDB用于缓冲写入的内存量（以字节为单位）。如果一个容器内有若干个任务实例，则每个容器中都有一个比例份额。此设置还可确定RocksDB的段文件的大小。 |
| stores.store-name.rocksdb.compression | 瞬间 |
此属性控制 RocksDB 是否应压缩磁盘上的数据和块缓存中的数据。以下值有效：
|
| stores.store-name.rocksdb.block.size.bytes | 4096 | 如果启用了压缩，RocksDB将这个许多未压缩字节分组到一个压缩块中。您可能不需要更改此属性。 |
| stores.store-name.rocksdb.ttl.ms | 商店的生存时间。请注意，它不是严格的TTL限制（仅在压实后才被删除）。请谨慎打开带有和不带TTL的数据库，因为它可能会损坏数据库。使用前请务必阅读约束。 | |
| stores.store-name.rocksdb.compaction.style | 普遍 |
该属性控制RocksDB在压缩其级别时将采用的压缩样式。以下值有效：
|
| stores.store-name.rocksdb.num.write.buffers | 3 | 配置 RocksDB 存储使用的写缓冲区数。这允许 RocksDB 继续对其他缓冲区进行写入，即使给定的写入缓冲区被刷新到磁盘。 |
| stores.store-name.rocksdb.max.log.file.size.bytes | 67108864 | RocksDB LOG 文件在旋转之前的最大大小（以字节为单位）。 |
| stores.store-name.rocksdb.keep.log.file.num | 2 | RocksDB LOG 文件的数量（包括旋转 LOG.old.*文件）要保留。 |
|
与集群管理一起运行 Samza | ||
| cluster-manager.container.memory.mb | 1024 | 每个容器的工作需要从集群管理器请求多少内存（兆字节）。与 cluster-manager.container.cpu.cores 一起，此属性确定集群管理器在一台计算机上运行的容器数量。如果容器超过这个限制，它将被杀死，所以重要的是容器的实际记忆使用量仍然低于极限。使用的内存量通常是JVM堆大小（配置为 task.opts），加上任何非堆内存分配的大小（例如 store.*.container.cache.size.bytes），另外还有一个安全余量以允许 JVM 开销。 |
| cluster-manager.container.cpu.cores | 1 | 每个容器的工作要求的CPU核心数量。集群中的每个节点都有一定数量的可用CPU核心，所以这个数量（以及 cluster-manager.container.memory.mb）决定了一台机器上可以运行多少个容器。 |
| cluster-manager.container.retry.count | 8 | 如果容器出现故障，Samza 会自动重新启动。但是，如果容器在启动后不久就会发生故障，这表示更深层的问题，所以我们应该杀死这个工作，而不是无限期地重试。此属性确定快速连续重新启动故障容器的最大次数（时间段配置为 cluster-manager.container.retry.window.ms）。作业中的每个容器都单独计数。如果此属性设置为0，任何失败的容器立即导致整个作业失败。如果设置为负数，则重试次数没有限制。 |
| cluster-manager.container.retry.window.ms | 300000 |
此属性确定容器在放弃并失败之前允许失败的频率。如果同一个容器的故障超过 cluster-manager.container.retry.count 次数，并且故障之间的时间小于此属性 cluster-manager.container.retry.window.ms（以毫秒为单位），则我们将失败。如果故障之间的时间大于，则我们将重新启动容器的次数没有限制cluster-manager.container.retry.window.ms。
|
| cluster-manager.jobcoordinator.jmx.enabled | 真 |
确定是否应该在作业的JobCoordinator上启动JMX服务器。（true或false）。
|
| cluster-manager.allocator.sleep.ms | 3600 | 容器分配器线程负责将请求与分配的容器进行匹配。此线程的睡眠间隔使用此属性进行配置。 |
| cluster-manager.container.request.timeout.ms | 5000 | 分配器线程会定期检查容器请求和分配的容器的状态，以确定容器对已分配资源的分配。此属性确定容器请求被认为已过期/超时之前的毫秒数。当请求过期时，它将被分配给集群管理器返回的任何可用容器。 |
|
在 YARN 集群上运行您的工作
（本节适用于您设置了 job.factory.class
= org.apache.samza.job.yarn.YarnJobFactory） | ||
| yarn.package.path |
YARN作业必需：可以从中下载作业包的 URL，例如http://或hdfs://URL。作业包是一个具有特定目录结构的 .tar.gz 文件 。
|
|
| yarn.container.memory.mb | 1024 | 这不支持 cluster-manager.container.memory.mb |
| yarn.container.cpu.cores | 1 | 这不支持 cluster-manager.container.cpu.cores |
| yarn.container.retry.count | 8 | 这不支持 cluster-manager.container.retry.count |
| yarn.container.retry.window.ms | 300000 | 这不支持 cluster-manager.container.retry.window.ms |
| yarn.am.container.memory.mb | 1024 | 在纱线中运行的每个Samza工作都有一个特殊的容器，即管理执行作业的ApplicationMaster（AM）。此属性确定从YARN请求运行ApplicationMaster的内存（兆字节）。 |
| yarn.am.opts | 任何JVM选项在执行Samza ApplicationMaster时都包含在命令行中 。例如，这可以用于设置 JVM 堆大小，调整垃圾回收器或启用远程调试。 | |
| yarn.am.java.home |
Samza AM 的 JAVA_HOME 路径。通过设置此属性，可以使用与集群的Java版本不同的java版本。记住设置task.java.home好。
|
|
| yarn.am.poll.interval.ms | 1000 | Samza ApplicationMaster会将定期的心跳发送到YARN ResourceManager以确认它是否存活。此属性确定心跳之间的时间（以毫秒为单位）。 |
| yarn.am.jmx.enabled | 真 | 这不支持 cluster-manager.jobcoordinator.jmx.enabled |
| yarn.allocator.sleep.ms | 3600 | 这不支持 cluster-manager.allocator.sleep.ms |
| yarn.samza.host-affinity.enabled | 假 | 这不支持 job.host-affinity.enabled |
| yarn.container.request.timeout.ms | 5000 | 这不支持 cluster-manager.container.request.timeout.ms |
| yarn.queue | 确定哪个 YARN 队列将用于 Samza 作业。 | |
| yarn.kerberos.principal | 当在启用 Kerberos 的 YARN 群集上运行时，Samza 作业的主体用于向 KDC 进行身份验证。 | |
| yarn.kerberos.keytab | 包含由主体指定的keytab的文件的完整路径，由yarn.kerberos.principal 指定。keytab 文件上传到 HDFS上每个应用程序唯一的登台目录，然后应用程序主机使用密钥表和主体定期登录以重新创建委托令。 | |
| yarn.token.renewal.interval.seconds | 应用程序主人重新认证和更新委托令牌的时间间隔。此值应小于授权令牌在有效期限之前在 hadoop namenode 上有效的时间长度。 | |
| yarn.resources.resource-name.path |
资源名称资源 本地化的路径。路径中的方案（例如http，ftp，hdsf，文件等）应在YARN core-site.xml中配置为fs。 |
|
| yarn.resources.resource-name.local.name | resource-name | 本地化后资源的新本地名称。此配置仅适用于yarn.resources.resource-name.path 已配置的时候。 |
| yarn.resources.resource-name.local.type | FILE |
本地化后资源的类型。它可以是 ARCHIVE（归档目录），FILE 或 PATTERN（使用模式从归档中提取的条目）。此配置仅适用于 yarn.resources.resource-name.path 已配置的时候。 |
| yarn.resources.resource-name.local.visibility | APPLICATION |
本地化后资源的可见性。它可以是 PUBLIC（对所有人都可见），PRIVATE（对于与此应用程序相同的帐户用户的所有Samza应用程序可见）或APPLICATION（仅对该Samza应用程序可见）。此配置仅适用于 yarn.resources.resource-name.path 已配置的时候。 |
| 指标 | ||
|
metrics.reporter.reporter-name.class
|
Samza 自动跟踪各种衡量指标，这些指标对于监控作业的健康状况非常有用，您还可以跟踪自己的指标。使用此属性，您可以定义任意数量的指标记录器，将指标发送到您选择的系统（用于绘图，警报等）。你给每个记者一个任意的记者名字。为了使记者能够参考记者名字，以 指标为准。此属性的值是实现 MetricsReporterFactory 的Java类的完全限定名称 。默认情况下，Samza会附带这些实现：
|
|
| metrics.reporters | 如果您已经使用 metrics.reporter.*.class 定义了任何指标记录器 ，那么您需要将它们列在这里才能启用它们。该属性的值是以逗号分隔的记者名称令牌列表。 | |
|
metrics.reporter.reporter-name.stream |
如果您已经注册了 metric metricsreports.export.*.class = org.apache.samza.metrics.reporter.MetricsSnapshotReporterFactory，那么您需要设置此属性来配置要发送度量数据的输出流。流以系统名称的形式给出 。流名称，系统必须在作业配置中定义。将许多不同的作业发布到相同的指标流可以很好。Samza定义了一个简单的 JSON编码 用于度量; 为了使用此编码，您还需要为度量流配置 serde：
|
|
| metrics.reporter.reporter-name.interval |
如果您已经注册了指标报告器 metricss.reporter.*.class = org.apache.samza.metrics.reporter.MetricsSnapshotReporterFactory，您可以使用此属性来配置记者报告注册的指标的频率。此属性的值应为连续指标报告间隔的长度。该值以秒为单位，应为正整数值。此属性是可选的，默认设置为60，这意味着每60秒报告一次。
|
|
| 写入HDFS | ||
| systems.system-name.producer.hdfs.writer.class | org.apache.samza.system.hdfs.writer.BinarySequenceFileHdfsWriter | 这个 HDFS Producer 系统应该使用的 HdfsWriter 的完全合格的类名 |
|
systems.system-name.
producer.hdfs.compression.type
|
没有 | 用于使用压缩类型的人类可读标签，例如“gzip”“snappy”等。根据 HdfsWriter 实现的性质，此标签将被不同地（或忽略）解释。 |
|
systems.system-name.
producer.hdfs.base.output.dir
|
/user/USERNAME/SYSTEMNAME | HDFS 的基本输出目录写入。默认为运行作业的用户的主目录，后跟 job.properties 文件中定义的此HdfsSystemProducer 的 systemName。 |
|
systems.system-name.
producer.hdfs.bucketer.class
|
org.apache.samza.system.hdfs.writer.JobNameDateTimeBucketer
|
用于管理 HDFS 路径和文件名的 Bucketer 实现的完全限定类名称。用于按时间批量写入或其他类似的分区方法。 |
|
systems.system-name.
producer.hdfs.bucketer.date.path.format
|
适用于 HDFS 路径的日期格式（使用 Java 的SimpleDataFormat 语法），可以配置输出文件的基于时间的功能。 | |
|
systems.system-name.
producer.hdfs.write.batch.size.bytes
|
268435456 | 在剪切新文件之前，要写入每个 HDFS 输出文件的传出消息的字节数。如果没有设置，默认为 256MB。 |
|
systems.system-name.
producer.hdfs.write.batch.size.records
|
262144 | 在剪切新文件之前要写入每个 HDFS 输出文件的传出消息的数量。如果未设置，默认为 262144。 |
| 从HDFS阅读 | ||
| systems.system-name.consumer.bufferCapacity | 10 | hdfs 用户缓冲区的容量 - 用于存储消息的阻塞队列。较大的缓冲区容量通常会导致更好的吞吐量，但会消耗更多的内存。 |
| systems.system-name.consumer.numMaxRetries | 10 | 在容器发生故障之前从 HDFS 获取消息失败时的重试尝试次数。 |
|
systems.system-name.
partitioner.defaultPartitioner.whitelist
|
* | 目录分区器使用的白名单，以 Java Pattern 样式在 hdfs 目录中选择文件。 |
|
systems.system-name.
partitioner.defaultPartitioner.blacklist
|
目录分区器使用的黑名单，以 Java Pattern 样式过滤 hdfs 目录中的不需要的文件。 | |
|
systems.system-name.
partitioner.defaultPartitioner.groupPattern
|
目录分区用于高级分区的组模式。高级分区超出了每个文件是分区的基本假设。使用高级分区，您可以任意地将文件分组到分区。例如，如果您有一组文件为：[part-01-a.avro，part-01-b.avro，part-02-a.avro，part-02-b.avro，part-03-a。 avro]， 你想组织分区：（part-01-a.avro，part-01-b.avro），（part-02-a.avro，part-02-b.avro），（part- 03-a.avro），其中中间的数字作为“组标识符”，您可以将此属性设置为 “part- [id] - .*”（注意 “[id]” 是一个保留的在这里，即你必须把它作为 “[id]”）。分区器将将此模式应用于所有文件名，并提取“组标识符”（“[id]” 在模式中），然后使用“组标识符”将文件分组到分区。查看更多详细信息 HdfsSystemConsumer设计文档（这是个下载链接） |
|
| systems.system-name.consumer.reader | avro | 用于不同事件格式的文件读取器类型（avro，plain，json等）。“avro” 现在只支持类型。 |
| systems.system-name.stagingDirectory | 用于存储分区描述的分段目录。默认情况下（如果不是由用户设置），该值将在内部从 “yarn.job.staging.directory” 继承。默认值通常足够好，除非您明确使用单独的位置。 | |
