Amoro运维管理 高可用 AMS支持“一主多备”的高可用性(HA)模式。该模式通过Zookeeper实现领导者选举机制,并需指定AMS集群名称及Zookeeper地址。AMS集群名称用于将同一Zookeeper集群下的不同AMS集群进行绑定,以避免相互干扰。 多个实例应连接同一个生产系统数据库,并保持配置、插件和 clustername 一致。 plaintext ams: ha: enabled: true clustername: production Differentiating binding multiple sets of AMS on the same ZooKeeper. zookeeperaddress: zk1:2181,zk2:2181,zk3:2181 ZooKeeper server address. 配置Flink Optimizer Flink Container通过Flink作业运行Optimizer,适用于在YARN或Kubernetes上扩展优化资源。配置分为运行环境、AMSContainer、OptimizerGroup和作业参数四部分。 部署前提 AMS节点已安装与集群兼容的Flink、Hadoop和YARN客户端,并能够使用目标提交用户访问HDFS、YARNResourceManager和HiveMetastore。 AMS与Optimizer节点时间同步;启用Kerberos时,krb5.conf、principal、keytab和Hadoop配置文件均可访问。 确认${AMOROHOME}/plugin/optimizer/flink/optimizerjob.jar存在,Flink版本与Amoro插件兼容。 确认以下环境变量已设置并生效(通常配置在/etc/profile.d/flink.sh文件下): plaintext export HADOOPHOME/usr/local/hadoop3 export FLINKHOME/usr/local/flink export HADOOPCONFDIR/etc/hadoop/conf export FLINKCONFDIR$FLINKHOME/conf export PATH$PATH:$FLINKHOME/bin:$HADOOPHOME/bin:$HADOOPCONFDIR export HADOOPCLASSPATHhadoop classpath 在配置文件添加 Flink Container 编辑 ${AMOROHOME}/conf/config.yaml(在翼MR Manager中Amoro的配置管理页面进行操作)。将 ams.serverexposehost 设置为 Optimizer 节点可访问的 AMS 主机名或 IP;不要使用 localhost。然后在 containers 中添加 Flink Container。完成配置项修改后,需重启amoro服务。 plaintext ams: serverexposehost: "X.X.X.X" containers: name: flinkContainer containerimpl: org.apache.amoro.server.manager.FlinkOptimizerContainer properties: flinkhome: /usr/local/flink target: yarnperjob export.HADOOPCONFDIR: /usr/local/hadoop3/etc/hadoop export.HADOOPUSERNAME: hdfs export.FLINKCONFDIR: /usr/local/flink/conf export.JVMARGS: Djava.security.krb5.conf/etc/krb5.conf 配置 Flink 运行参数 根据集群环境修改 ${FLINKHOME}/conf/flinkconf.yaml(在翼MR Manager中Flink的配置管理页面进行操作)。Kerberos 集群至少应核对 security.kerberos.login.useticketcache、security.kerberos.login.keytab、security.kerberos.login.principal 和 security.kerberos.login.contexts等参数,示例配置: plaintext security.kerberos.login.useticketcache: true security.kerberos.login.keytab: /etc/security/keytabs/hdfs.keytab security.kerberos.login.principal: hdfs/ @ security.kerberos.login.contexts: Client,KafkaClient 完成配置项修改后,需保存并同步配置到Flink节点(具体操作见++配置服务参数++)。 YARN 场景应确认队列、日志聚合、Hadoop 配置目录和 Java 路径。以下参数也可在 Optimizer Group 的 Properties 中以 flinkconf. 形式覆盖。 配置项 建议值 说明 flinkconf.yarn.classpath.includeuserjar DISABLED 与 childfirst 配合,降低 Amoro 与 Flink lib 中 Iceberg 版本冲突风险。 flinkconf.classloader.resolveorder childfirst 优先加载 Optimizer 作业携带的依赖。 flinkconf.taskmanager.memory.managed.size 32mb Optimizer 不使用 Flink 状态,托管内存可按压测结果调小。 flinkconf.taskmanager.memory.network.min 32mb Optimizer 算子间网络开销较小,可降低网络内存下限。 flinkconf.taskmanager.memory.network.max 32mb 与 network.min 保持一致;大规模任务需结合运行指标复核。 flinkconf.taskmanager.memory.process.size 按数据量设置 内存不足时逐步增大,例如从 1728m 或 2g 起进行压测。 flinkconf.taskmanager.numberOfTaskSlots 1 单槽隔离便于控制单个 Optimizer 的资源;提高前先评估内存与并发。 创建 Optimizer Group 登录 WebUI,进入 Optimizing > Optimizer Groups,新增 Group 并将 Container 选择为 flinkContainer。Properties 是 Scale Out 时的默认作业参数;参数优先级为 Optimizer Group > Flink Container > flinkconf.yaml。 属性 是否必填 说明 schedulingpolicy 否 quota 按表配额调度;balanced 优先调度等待时间较长的表。默认 quota。 amsoptimizinguri 否 表优化服务端点。当默认端点不可访问时,配置 thrift:// :1261。 flinkconf. 否 覆盖任意 Flink 原生配置,例如内存、队列、并行度和类加载参数。 关联 Catalog 并启动 1、在Catalog 配置中将Optimizer Group指向新建的Flink Group,并确认目标表已启用自优化。 2、返回Optimizer Groups,单击Scale Out,设置并行度后确认。提交到YARN可能需要数分钟。 3、在Optimizers页面确认实例状态为RUNNING;同时在 YARN Application中确认作业已启动。 4、观察Pending、Executing、Committing、失败数和 TaskManager内存;确认至少一个完整优化周期能够正常完成。 运维注意事项 Scale Out无响应时,先检查ams.serverexposehost、HADOOPCLASSPATH、HADOOPCONFDIR、YARN 客户端和 1261 端口连通性,其次查看日志是否有异常报错。 出现No Executor Found时,重点检查 HADOOPCLASSPATH是否包含 hadoop classpath,以及提交账户是否能访问Hadoop/YARN配置。 删除Optimizer Group前,先解除Catalog关联并Release组内Optimizer;若外部 Application未退出,应在确认无任务提交后到YARN侧终止。 不要将真实 principal、keytab 内容、密码或集群地址写入对外文档;示例值应替换为实际环境配置,并妥善保护凭据。