Apache Samza 1.8.0 配置参数详解

Apache Samza 1.8.0 配置参数详解：覆盖 properties 修改与优先级、ZK/Kafka 连接、JVM 堆与进程资源（256-512MB）、序列化器与 KV 存储及生产模板，键名均实测验证

Apache SamzaApache Samza2026/10/13

Apache Samza 1.8.0 配置参数详解

目标：详解 Apache Samza 1.8.0 作业配置（properties）的核心参数，覆盖配置加载方式、参数优先级、ZooKeeper/Kafka 连接、JVM 堆与进程资源、序列化器与 KV 存储，并给出本地运行与 YARN 模式两套配置模板。以下参数与键名均在 Ubuntu 22.04 真实环境逐步实测验证。

一、环境信息

项目值
Apache Samza1.8.0（Scala 2.12）
配置载体作业 properties 文件（Java properties 格式）
样例配置文件/opt/hello-samza/deploy/samza/config/wikipedia-parser.properties、wikipedia-stats.properties、wikipedia-application-local-runner.properties
配置加载器org.apache.samza.config.loaders.PropertiesConfigLoaderFactory
依赖组件ZooKeeper 3.9.6（2181）、Kafka 4.3.1（9092）、Hadoop 3.4.1（YARN 模式）

二、前置条件

  • Samza 作业依赖 ZooKeeper 与 Kafka，YARN 模式另需 Hadoop YARN（前置部署见《Apache Samza 1.8.0 安装指南》）。
  • 配置文件为 Java properties 格式，键值用 = 连接，每行一个键；作业运行前由 --config-path= 或 job.config.loader.properties.path 指定。

三、参数查看与修改方式

3.1 配置加载入口

run-app.sh 通过 --config-path 指定 properties：

cd /opt/hello-samza/deploy/samza
./bin/run-app.sh --config-path=config/wikipedia-parser.properties

本地运行模式通过 --config 覆盖指定加载器与文件路径：

./bin/run-class.sh samza.examples.wikipedia.application.WikipediaZkLocalApplication \
  --config job.config.loader.factory=org.apache.samza.config.loaders.PropertiesConfigLoaderFactory \
  --config job.config.loader.properties.path=config/wikipedia-application-local-runner.properties

3.2 参数优先级（实测）

  1. 命令行 --config key=value 覆盖（最高）；
  2. properties 文件内容（job.config.loader.properties.path 指向）；
  3. 应用代码中描述符（Descriptor）声明的默认值（最低）。

实测证明：properties 中设置 systems.kafka.producer.bootstrap.servers=localhost:9999（错误端口）后，作业生产者按 9999 连接失败，证明 properties 覆盖描述符默认值生效。

3.3 运行时核对生效值

# JVM 堆生效值
ps -p <pid> -o cmd | grep -oE '\-Xms[0-9]+m \-Xmx[0-9]+m'

# 作业日志中打印的完整配置
grep -E 'bootstrap.servers|zookeeper.connect' $SAMZA_LOG_DIR/stdout

四、参数总览

参数默认值说明
job.name（必填）作业名，用于 ZK 节点与 YARN 应用名
app.class（必填）TaskApplication 实现类
job.factory.classorg.apache.samza.job.yarn.YarnJobFactory作业工厂（YARN 模式）
job.coordinator.factoryorg.apache.samza.zk.ZkJobCoordinatorFactory协调器工厂（ZK 模式）
job.coordinator.zk.connect（必填）协调器 ZooKeeper 地址
job.coordinator.systemkafka协调器消息系统名
job.default.system（必填）默认系统
job.config.loader.factoryPropertiesConfigLoaderFactory配置加载器
job.config.loader.properties.path（必填）properties 文件路径
systems.<id>.samza.factoryKafkaSystemFactory系统工厂类
systems.kafka.consumer.zookeeper.connectlocalhost:2181/消费者位移 ZooKeeper
systems.kafka.producer.bootstrap.serverslocalhost:9092生产端 Kafka 地址
yarn.am.container.memory.mb512AM 进程内存（MB）
yarn.am.opts-Xms256m -Xmx512mAM JVM 参数
cluster-manager.container.memory.mb512task 进程内存（MB）
task.opts-Xms256m -Xmx512mtask JVM 参数
serializers.registry.<id>.class无序列化器工厂类
stores.<name>.factoryRocksDbKeyValueStorageEngineFactoryKV 存储工厂
stores.<name>.changelog无changelog topic
stores.<name>.key.serde / msg.serde无键/值序列化器
task.name.grouper.factoryGroupByContainerIdsFactorytask 分组策略

五、核心参数分类详解

5.1 作业与运行模式

job.name=wikipedia-parser
app.class=samza.examples.wikipedia.task.application.WikipediaParserTaskApplication
job.factory.class=org.apache.samza.job.yarn.YarnJobFactory
job.config.loader.factory=org.apache.samza.config.loaders.PropertiesConfigLoaderFactory
job.config.loader.properties.path=./__package/config/wikipedia-parser.properties
  • app.class：TaskApplication 实现类，Samza 通过反射实例化并构建任务图；
  • job.factory.class：YARN 模式为 org.apache.samza.job.yarn.YarnJobFactory；本地模式不设置（LocalApplicationRunner 内部处理）；
  • job.name 为旧式键，日志会出现 job.name is a deprecated configuration, use use app.name instead. 警告，仍可用。

5.2 协调器与 ZooKeeper

job.coordinator.factory=org.apache.samza.zk.ZkJobCoordinatorFactory
job.coordinator.zk.connect=zookeeper:2181
job.coordinator.system=kafka
  • job.coordinator.factory：ZK 协调器负责 leader 选举、任务分配与 jobModel 管理；
  • job.coordinator.zk.connect：单机 host:port，集群用逗号分隔 h1:2181,h2:2181,h3:2181；
  • 协调器相关 ZK 节点根路径为 /app-<job.name>-1（如 /app-wikipedia-application-compose-1）。

5.3 Kafka 系统配置

systems.kafka.samza.factory=org.apache.samza.system.kafka.KafkaSystemFactory
systems.kafka.consumer.zookeeper.connect=zookeeper:2181/
systems.kafka.producer.bootstrap.servers=kafka:9092
  • 系统标识 kafka 需与 job.default.system=kafka 一致；
  • consumer.zookeeper.connect：Samza 消费者位移存储在 ZooKeeper，路径末位 / 不可省；
  • producer.bootstrap.servers：生产者直连 Kafka broker，Compose 环境用服务名，物理机用 host:port 逗号列表；
  • 注意：Samza 的 systems.<id>.consumer.zookeeper.connect 与 systems.<id>.producer.bootstrap.servers 是 Samza 专用键，不要与 Kafka 原生 consumer 配置混淆。

5.4 JVM 堆与进程资源（256-512MB）

yarn.am.container.memory.mb=512
yarn.am.opts=-Xms256m -Xmx512m
cluster-manager.container.memory.mb=512
task.opts=-Xms256m -Xmx512m
  • yarn.am.container.memory.mb：ApplicationMaster 进程内存请求（实测提交日志 set memory request to 512）；
  • yarn.am.opts：AM JVM 参数，堆限制 256-512MB；
  • cluster-manager.container.memory.mb：task 进程内存请求；
  • task.opts：task JVM 参数；
  • 实测生效：coordinator 与 task 进程命令行均含 -Xms256m -Xmx512m；提交日志确认 set cpu core request to 1（vCore 1）。

5.5 序列化器与 KV 存储

serializers.registry.string.class=org.apache.samza.serializers.StringSerdeFactory
serializers.registry.integer.class=org.apache.samza.serializers.IntegerSerdeFactory
stores.wikipedia-stats.factory=org.apache.samza.storage.kv.RocksDbKeyValueStorageEngineFactory
stores.wikipedia-stats.changelog=kafka.wikipedia-stats-changelog
stores.wikipedia-stats.key.serde=string
stores.wikipedia-stats.msg.serde=integer
  • serializers.registry.<id>.class：注册序列化器工厂，<id> 为后续引用的短名（string/integer）；
  • stores.<name>.factory：KV 存储实现，默认 RocksDB（RocksDbKeyValueStorageEngineFactory），也可用 InMemory；
  • stores.<name>.changelog=kafka.<topic>：changelog topic 需前缀 kafka.（系统名），由作业自动创建；
  • key.serde / msg.serde：引用 serializers.registry 注册的短名；
  • ⚠️ 若设置系统级 systems.<id>.samza.msg.serde=json 而未定义 serializers.json.class，作业报 buildSystemSerdeMap: No class defined for serde: json 并停止（见常见问题）。

5.6 task 分组

task.name.grouper.factory=org.apache.samza.container.grouper.task.GroupByContainerIdsFactory

单分区场景使用默认分组器即可；多分区场景按分组策略决定各 task 归属。

六、生产配置模板

6.1 本地运行模式模板

job.name=wikipedia-application
job.coordinator.factory=org.apache.samza.zk.ZkJobCoordinatorFactory
job.coordinator.zk.connect=localhost:2181
job.coordinator.system=kafka
job.default.system=kafka
task.name.grouper.factory=org.apache.samza.container.grouper.task.GroupByContainerIdsFactory
serializers.registry.string.class=org.apache.samza.serializers.StringSerdeFactory
serializers.registry.integer.class=org.apache.samza.serializers.IntegerSerdeFactory
stores.wikipedia-stats.factory=org.apache.samza.storage.kv.RocksDbKeyValueStorageEngineFactory
stores.wikipedia-stats.changelog=kafka.wikipedia-stats-changelog
stores.wikipedia-stats.key.serde=string
stores.wikipedia-stats.msg.serde=integer
systems.kafka.samza.factory=org.apache.samza.system.kafka.KafkaSystemFactory
systems.kafka.consumer.zookeeper.connect=localhost:2181/
systems.kafka.producer.bootstrap.servers=localhost:9092

6.2 YARN 模式模板

# Job
job.factory.class=org.apache.samza.job.yarn.YarnJobFactory
job.name=wikipedia-parser

# TaskApplication class
app.class=samza.examples.wikipedia.task.application.WikipediaParserTaskApplication

# Config Loader
job.config.loader.factory=org.apache.samza.config.loaders.PropertiesConfigLoaderFactory
job.config.loader.properties.path=./__package/config/wikipedia-parser.properties

# Container resources: 512MB per process, JVM heap 256-512MB
yarn.am.container.memory.mb=512
yarn.am.opts=-Xms256m -Xmx512m
cluster-manager.container.memory.mb=512
task.opts=-Xms256m -Xmx512m

💡 YARN 模式还需在提交端设置 HADOOP_CONF_DIR=/opt/hadoop/etc/hadoop，并在配置中指定 yarn.package.path（dist tar 路径，如 file:///opt/hello-samza/target/hello-samza-1.6.0-dist.tar.gz）。

七、常见问题

7.1 作业报 buildSystemSerdeMap: No class defined for serde: json

错误回显（日志）：

ScheduleAfterDebounceTime [ERROR] Execution of action: BarrierAction failed.
org.apache.samza.SamzaException: buildSystemSerdeMap: No class defined for serde: json.
ZkJobCoordinator [ERROR] Received exception in debounce timer! Stopping the job coordinator

定位：配置了 systems.<id>.samza.msg.serde=json 等系统级 serde，但 serializers.registry.json.class 未注册。

解决：删除系统级 serde 行，或补注册：serializers.registry.json.class=org.apache.samza.serializers.JsonSerdeFactory。

7.2 run-app.sh 无参报 ConfigException

错误回显：

org.apache.samza.config.ConfigException: Legacy task applications must set a non-empty task.class in configuration.

定位：未指定 --config-path，properties 未加载。

解决：./bin/run-app.sh --config-path=config/wikipedia-parser.properties。

7.3 提交端报 RM 连接失败

错误回显：

ClientHelper [WARN] Trying to connect to RM localhost:8032 failed

定位：HADOOP_CONF_DIR 未指向真实 Hadoop 配置目录（Samza 脚本默认 $HOME/.samza/conf 不存在），或 RM 未启动。

解决：export HADOOP_CONF_DIR=/opt/hadoop/etc/hadoop 后重试；确认 RM 8088/8032 端口开放。

7.4 kill-yarn-job.sh 传 --config-path 报 Invalid ApplicationId prefix

定位：脚本按位置参数解析 AppId，--config-path= 被误判为 AppId。

解决：直接传 AppId：./bin/kill-yarn-job.sh application_1791822120187_0007。

八、参考链接