思科基于 Amoro + Apache Iceberg 构建云原生湖仓实践

Amoro 是一个构建在 Apache Iceberg 等开放数据湖表格之上的湖仓管理系统，提供了一套可插拔的数据自优化机制和管理服务，旨在为用户带来开箱即用的湖仓使用体验。

01 作者简介

白旭，来自思科 WebEx 数据平台的 Software Engineer，主要负责数据湖仓一体的研发与优化。

02 Amoro 在 Webex

Cisco Webex products provide capabilities including online meetings, team messaging and file sharing. The suite is considered a leading collaboration platform in the unified communications area and is geared toward both small group collaboration for SMBs as well as large group meetings for enterprise-wide deployments.

为什么选择 Amoro

Webex 大数据团队最初采用 Hive 存储格式作为主要标准。然而，由于 Hive 存储格式的限制，纠正特定客户数据和进行数据回溯都变得非常低效。此外，使用 Hive 也增加了维护开销。鉴于这些挑战，我们着手寻找替代的存储解决方案。这时，我们发现了 Apache Iceberg，它的设计不仅能够降低我们的运营和维护成本，还能提高核心业务运营的效率。因此，我们着手构建基于 Apache Iceberg 的数据湖。

去年开始 Webex 逐渐使用 Iceberg V2 format 来当作默认的表格式。V2 的重要特性之一就是 row-level 的更新能力，但是引入了关键的难题：读取 V2 表时Merge-on-Read（MOR）带来的性能问题。如果累计了大量的 Delete 文件，依赖 Iceberg 的报表查询的时效性将大打折扣，甚至达到了不可查的情况。

我们最先开始尝试利用 Spark comapction procedures 来合并小文件，通过调度系统定时依次执行。

但是这样相对原始的维护手段有诸多弊端：

● 高资源占用：每个Spark job可能会超过40core和300GB的内存

● 执行时间长：每个环境有诸多Iceberg表需要合并，并且当合并数据量大的表时会阻塞其他表，这使得每个Pocedure执行时间都过长

● 低容错：一个合并任务可能包含多个Iceberg表，当某个表出现错误时会导致整个Job的失败

● 维护困难：合并失败的表需要不断重启，手动修复，甚至可能会被遗漏

正是有上述多个痛点，我们很快引入了 Amoro 来解决 Iceberg 表的诸多维护问题。我们采用 External Flink Optimizer 注册的方式，拉取并处理 Amoro Management Service（AMS）生产的优化任务。并且开启了快照过期，数据过期等服务来减少存储系统的压力。

Amoro 解决了传统定时调度的 Spark job 的诸多痛点并带来了多个优势：

● 更高的资源利用率：单从资源占用最多的环境来看，使用 Flink Optimizer 节省了 70% 左右的资源使用

● 高容错：失败的 Optimization process/task 可以在下次扫描后重新尝试优化

● 及时性：持续的合并优化可以保证 Iceberg 表的查询效率维持在可控范围，保证报表的查询效率

● 自管理：Amoro 仅根据表的 properties 来确定是否开启了优化，可以控制表级的优化开启或关闭和相关策略

● 可视化：Amoro 提供 WebUI，将优化状态，表的基本信息等情况呈现给开发者

使用情况

Cisco Webex 在 Amoro（前Arctic）项目开源后不久就尝试用其来解决 Iceberg 的小文件合并，快照过期清理等数据湖治理的各类问题。直至今日，Amoro 已在 Webex 上有了一定的实践规模：

● 多数据中心，多集群部署：最多7个不同数据中心，Hadoop 集群环境

● 多个环境：我们不仅在 Hadoop 环境中利用 Amoro 管理 Iceberg 表，近期将其部署在 AWS 环境并优化 AWS 上 Iceberg 表

● 1000+ Iceberg 表

不过根据公司的实际情况，我们会逐步将 Hadoop 环境迁移至 AWS 环境，在此期间遇到的实际问题和解决方案可以在此篇文章做些讨论。

03 Amoro on AWS

AWS 上需要解决多个问题，首先的是 Iceberg AWS 集成如 Catalog 和 FileSystem 的切换等，另外就是 AMS 端的适配。

Iceberg AWS Integrations

我们将 Iceberg 上线到 AWS 后，从 HiveCatalog 切换至 GlueCatalog，并且也将文件系统由 HDFS 切换为 S3。相对于 HDFS，S3 有着更完善的权限控制，可以给不同 IAM 账号划分 bucket，甚至文件的不同权限，来实现细粒度的权限控制。这是确保数据湖的数据安全性和隐私的关键要素。其次，HDFS 通常需要自行管理硬件和维护，因此可能需要更多的初始成本和运营成本，S3 是云服务，其成本模型是按使用量计费，没有硬件维护成本。

Hive Metastore 作为服务在不同的环境需要各自部署，同时也需要维护 MySQL 来存储元数据信息，相比于 Glue 增加了自身维护成本。比如我们在实际的生产环境中就遇到因为 MySQL 连接数达到上限而不得重启 HMS 的情况。而 Glue 相对 HMS 更简单，不需要手动维护，也减少多个环境分开部署导致的数据孤岛问题。

LockManager

对于 Iceberg 来说，有些文件系统如 S3 不提供文件的写入排斥来保证 metadata 原子性，HMS 依赖 MySQL 做了排他锁，而 Glue 需要自定义锁的实现，在  AWS 上 Iceberg 提供 DynamoDBLockManager 保证表的并发修改。

1.  Iceberg 修改新 metadata.json 文件前向 LockManager 服务尝试获取锁

2.  如果有其他进程拿到锁时会重新尝试，这个尝试的次数和间隔均有相关参数可以配置

3.  拿到锁后将当前 metadata 文件的位置替换为新写入的 metadata-v2.json 地址

4.  如果在多次尝试后仍无法获取锁，那么此次提交可能会失败，然后再次重试

5.  成功提交 metadata.json 文件后最终 Iceberg 释放掉锁

DynamoDB lock table 其数据模型如下：

● entityId 是 DynamoDB 表中的 Key，由 Iceberg 数据库名称和表名组成

● lease duration 是锁心跳超时时间，如果锁超时会自动释放

● version 是一串随机由 DynomoDBHeartbeat 更新的 UUID，确保锁被当前线程持有

● ownerId 是待写入的新 metadata.json 文件

使用 DynamoDB 管理 Iceberg 表的并发修改，可以避免出现 Hive Metastore Service 中锁未释放的导致 job block，因为 DynamoDB 中如果拿锁的进程异常退出，其锁也会因为 lease 到期后自动释放，无需手动解锁。

权限控制

AWS IAM 账户可以对 S3，Glue，DynamoDB 等相关服务做细粒度的权限控制。当前我们给每个团队划分一个 IAM 账号，IAM account 授予相关 S3 bucket 的读写权限，同时也可以管理 Glue 的权限。我们的 AMS 服务和读写 Job 都是运行在 Kubernetes 上，因此可以天然使用 Namespace 作为单位来分配 IAM 账号，管理好每个人的相关 namespace 的权限，进而可以简单实现 Iceberg 表的权限控制，后面也会对 S3 和 Glue 权限做更详细的划分。

S3 Intelligent-Tiering

我们 Webex team 在调研 S3 读写 Iceberg 成本的过程中发现 AWS S3 中的一项属性比较有价值，但是 Iceberg 中暂未适配：storage-class。将该配置设置为 S3 Intelligent-Tiering 可以根据不同访问频率优化访问成本：

Amazon S3 Intelligent-Tiering 存储类旨在通过当访问模式改变时自动将数据移动到最具成本效益的访问层来优化存储成本。S3 Intelligent-Tiering 存储类自动将对象存储在三个访问层中：一个针对频繁访问进行了优化的层，一个针对不频繁访问进行了优化的更低成本的层，以及一个针对很少访问的数据优化的极低成本层。每月只需支付少量的对象监控和自动化费用，S3 Intelligent-Tiering 即可将连续 30 天未访问的对象移动到不频繁访问层，实现 40% 的节省，并在 90 天未访问之后，将其移动到归档即时访问层，实现 68% 的节省。

该修改已贡献给 Iceberg 社区，并在1.4.0中发布

https://github.com/apache/iceberg/releases/tag/apache-iceberg-1.4.0

AMS AWS Adaptions

早期 AMS 为了运行在 AWS 环境，通过 Custom Catalog 的方式创建并适配 Iceberg 的 GlueCatalog，并且重构了 Arctic 的 FileIO 相关接口以支持对象文件系统。预计 0.6.0 版本将 GlueCatalog 作为单独的 Catalog 类型，例如要求填写 IAM 等相关属性，会对 AWS 环境会有更好的适配。

无论是 AMS 还是 Optimizer 都需要访问 Iceberg 及其文件，直接的方式是给上述组件赋予能够访问所需合并表读写权限的 IAM account，我们是将 IAM 相关的认证放在环境变量中，默认的读取链 DefaultAWSCredentialsProviderChain 会正确拿到相关认证信息完成鉴权。在 k8s 环境中放在 env 变量中即可，例如：

apiVersion: apps/v1kind: Deploymentmetadata: labels: app.kubernetes.io/name: ams name: amsspec: ... template: metadata: labels: app.kubernetes.io/name: ams spec: ... containers: - env: - name: AWS_ACCESS_KEY_ID value: AKIXXXXXXXXXXXXXXXX - name: AWS_SECRET_ACCESS_KEY              value: fjHyrM1wTJ8CLP13+GU1bCGG1RGlL1xT1lXyBb11

此外，AWS 在2019年引入 IAM Roles for Service Accounts（IRSA），利用 AWS Identity API、OpenID Connect（OIDC）和 Kubernetes Service Accounts 对 Kubernetes pod 应用细粒度访问控制。IAM roles 与 Service Accounts 相绑定，相较于配置 AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY 等参数更简洁，也更安全，不会直接暴露认证明文信息。

IRSA ：https://aws.amazon.com/cn/blogs/containers/diving-into-iam-roles-for-service-accounts/

04 部署实践

此篇文章我们基于 Amoro 0.6.0 的版本作为示例，大致梳理以 Helm charts 为模版在 Kubernetes 环境上部署 Amoro 和开启 Iceberg 优化的流程。利用 Helm Chart 部署相对简略，更加详细的上云部署方式可以参考之前的文章：Apache Iceberg + Arctic 构建云原生湖仓实战。

值得注意的是，我们对 Amoro 及 Helm 做了些许改动，因此会与 Amoro 社区的 Helm charts 略有不同，但大致流程是一致的。

打包镜像

手动打包：

mvn clean install -DskipTests -am -e -pl dist

之后将打包好的 zip 包放入带有 Dockerfile 的目录中并打包 docker image:

docker build docker/ams/ --platform amd64 -t xxx/amoro && docker push xxx/amoro

打包好的 docker image 会在下述 Deployment 资源中引用并部署。

编写 Helm chart

除了基本的配置信息会放入Values中以外，带有保密属性的 IAM 认证信息， Database 的账户密码都会存入 Secrets。将 AMS 的配置信息如 bin/config.sh ，conf/config.yaml 抽出作为 ConfigMap 分别挂载至 env 和 volumeMounts。限于篇幅原因不会在此全部列出配置文件。

● _helpers.tpl 模版，与预定义了镜像，label 等基础信息

{{- define "udp.amoro.image.fullname" -}}

{{ .Values.image.repository }}/{{ .Values.image.component }}:{{ .Values.image.tag | default .Chart.AppVersion }}

{{- end -}}

{{- define "udp.amoro.common.labels" -}}

app.kubernetes.io/instance: {{ .Release.Name }}

app.kubernetes.io/managed-by: {{ .Release.Service }}

{{- end -}}

{{- define "amoro.home" -}}

{{ .Values.amoroHome | default "/usr/local/amoro" }}

{{- end -}}

● _pod.tpl 模版，定义 pod 挂载的配置文件等

{{- define "amoro.pod.container.mounts" }}

- name: logs

mountPath: {{ include "amoro.home" . }}/logs

- name: conf

mountPath: {{ include "amoro.home" . }}/conf/config.yaml

readOnly: true

subPath: "config.yaml"

{{- if or .Values.amoroConf.log4j2 }}

{{- /* log4j2.yaml from config-map*/ -}}

- name: conf

mountPath: {{ include "amoro.home" . }}/conf/log4j2.xml

readOnly: true

subPath: "log4j2.xml"

{{- end }}

{{- if or .Values.jvmOptions }}

- name: conf

mountPath: {{ include "amoro.home" . }}/conf/jvm.properties

readOnly: true

subPath: "jvm.properties"

{{- end -}}

{{- end -}}

{{- /* define amoro.pod.container.mounts end */ -}}

{{/* defined volumes for pod */}}

{{- define "amoro.pod.volumes" -}}

- name: conf

configMap:

name: config.yaml

- name: logs

emptyDir: {}

{{- end -}}

{{- /* define "amoro.pod.volumes" end */ -}}

● Deployment

与社区不同的是，我们通过IRSA或IAM认证放入环境变量中的方式进行进行统一管理和更新；另外Optimizer采用的是external模式，更关注AMS和Optimizer之间的连通性，所以把livenessProbe探针改为optimizing的TCP端口。

apiVersion: apps/v1

kind: Deployment

metadata:

labels:

app.kubernetes.io/name: ams

name: ams

spec:

replicas: {{ .Values.replicas }}

selector:

matchLabels:

app.kubernetes.io/name: ams

strategy:

type: {{ .Values.strategy.type | quote }}

rollingUpdate:

maxSurge: {{ .Values.strategy.rollingUpdate.maxSurge | quote }}

maxUnavailable: {{ .Values.strategy.rollingUpdate.maxUnavailable | quote }}

template:

metadata:

labels:

app.kubernetes.io/name: ams

spec:

{{- if .Values.affinity }}

affinity:

{{- toYaml .Values.affinity | nindent 8 }}

{{- end }}

{{- if .Values.nodeSelector }}

nodeSelector:

{{- toYaml .Values.nodeSelector | nindent 8 }}

{{- end }}

{{- if .Values.tolerations }}

tolerations:

{{- toYaml .Values.tolerations | nindent 8 }}

{{- end }}

{{- if .Values.image.pullSecret }}

imagePullSecrets:

- name: {{ .Values.image.pullSecret }}

{{- end }}

serviceAccountName: {{ .Values.serviceAccount.name }}

containers:

- env:

- name: AMS_DATABASE_PASSWORD

valueFrom:

secretKeyRef:

name: udp-amoro-externaldb-secret

key: database-password-udp

image: {{ include "udp.amoro.image.fullname" .}}

imagePullPolicy: {{ .Values.image.pullPolicy }}

name: ams

ports:

- containerPort: {{ .Values.ports.amoroServer }}

name: "amoro-server"

- containerPort: {{ .Values.ports.optimizing }}

name: "optimizing"

- containerPort: {{ .Values.ports.jmxExporter }}

name: "jmx-exporter"

{{- if .Values.livenessProbe.enabled }}

livenessProbe:

tcpSocket:

port: {{ .Values.ports.optimizing }}

initialDelaySeconds: {{ .Values.livenessProbe.initialDelaySeconds }}

periodSeconds: {{ .Values.livenessProbe.periodSeconds }}

timeoutSeconds: {{ .Values.livenessProbe.timeoutSeconds }}

failureThreshold: {{ .Values.livenessProbe.failureThreshold }}

successThreshold: {{ .Values.livenessProbe.successThreshold }}

{{- end }}

{{- if .Values.readinessProbe.enabled }}

readinessProbe:

httpGet:

path: /versionInfo

port: amoro-server

initialDelaySeconds: {{ .Values.readinessProbe.initialDelaySeconds }}

periodSeconds: {{ .Values.readinessProbe.periodSeconds }}

timeoutSeconds: {{ .Values.readinessProbe.timeoutSeconds }}

failureThreshold: {{ .Values.readinessProbe.failureThreshold }}

successThreshold: {{ .Values.readinessProbe.successThreshold }}

{{- end }}

resources:

{{- toYaml .Values.resources | nindent 12 }}

volumeMounts:

{{- include "amoro.pod.container.mounts" . | nindent 12}}

securityContext:

{{- toYaml .Values.podSecurityContext | nindent 12 }}

dnsPolicy: ClusterFirst

restartPolicy: {{ .Values.image.restartPolicy }}

schedulerName: default-scheduler

terminationGracePeriodSeconds: 30

volumes:

        {{- include "amoro.pod.volumes" . | nindent 8}}

● Service

定义Amoro Pod访问的端口

apiVersion: v1

kind: Service

metadata:

labels:

{{- include "udp.amoro.labels" . | nindent 4 }}

name: ams

spec:

ports:

- name: amoro-server

port: {{ .Values.ports.amoroServer }}

protocol: TCP

- name: optimizing

port: {{ .Values.ports.optimizing }}

protocol: TCP

- name: jmx-exporter

port: {{ .Values.ports.jmxExporter }}

protocol: TCP

selector:

    {{- include "udp.amoro.labels" . | nindent 4 }}

● ServiceAccount

{{- if .Values.serviceAccount.create -}}

apiVersion: v1

kind: ServiceAccount

metadata:

name: {{ .Values.serviceAccount.name }}

labels:

{{- include "udp.amoro.labels" . | nindent 4 }}

annotations:

eks.amazonaws.com/role-arn: {{ .Values.serviceAccount.iamRoleARN }}

{{- end -}}

● Secert

我们结合 SecertStore，ExternalSecret 管理敏感信息，将其存储在外部系统（如 Vault,Azure Key Vault 等）中，并通过引用这些外部密钥和凭据将其注入 Kubernetes 的 Secert 对象中。

● Ingress

Ingress 管理 Kubernetes 集群内部的 HTTP,HTTPS 路由，将外部流量路由到集群内部的服务，所以配置 Ingress 后我们可以通过 host 的方式访问其 WebUI

{{- if .Values.ingress.enabled -}}

apiVersion: networking.k8s.io/v1

kind: Ingress

metadata:

name: amoro

labels:

{{- include "udp.amoro.labels" . | nindent 4 }}

{{- with .Values.ingress.annotations }}

annotations:

{{- toYaml . | nindent 4 }}

{{- end }}

spec:

rules:

- host: {{ .Values.ingress.host }}

http:

paths:

- path: {{ .Values.ingress.path }}

pathType: {{ .Values.ingress.pathType }}

backend:

service:

name: ams

port:

number: {{ .Values.ports.amoroServer }}

{{- end }}

● Podmonitor

为了方便监控 AMS 的状态，我们在 Docker 镜像中配置 Prometheus 并暴露 jmx exporter 端口，以便收集和存储与 Amoro Pod 相关的度量指标。

apiVersion: monitoring.coreos.com/v1

kind: PodMonitor

metadata:

labels:

app.kubernetes.io/name: amoro-monitor

name: amoro-monitor

spec:

namespaceSelector:

any: true

podMetricsEndpoints:

- interval: 60s

port: jmx-exporter

selector:

matchLabels:

      app.kubernetes.io/name: ams

编写完 Helm 模版后通过如下命令安装部署

-- 安装/升级amoro helm

helm upgrade -install amoro ./ --namespace amoro

注册 GlueCatalog

● warehouse 为必填项，指定数据仓库根目录，当然你也可以使用简易的 warehouse 地址作为占位

● 在生产环境中配置建议配置 lock-impl 和 lock.table 用于保证 metadata.json 文件的修改原子性

● 对于使用 IRSA 认证的 AMS，需要设置参数 client.credentials-provider 为 software.amazon.awssdk.auth.credentials.WebIdentityTokenFileCredentialsProvider 来获取正确的认证信息

05 未来规划

1. 增量 SORT/ZORDER SORT：Data skipping 对于 Iceberg 的查询效率提升是显著的，尤其是对于查询条件相对固定的报表。此外更高效的文件跳过也可以减少读取表时对文件的 IO，降低外部访问 S3 时造成的流量开销。Clustering 的优化方式也让针对不同表的智能优化成为可能。

2. 完善监控告警：当前会基于 Amoro 的相关记录和指标做监控，例如 optimizing/pending/committing 超时告警，pod 状态监控等。除了这些还需要基于表本身的状态做健康状态的监控，提前预报表的读/写放大。

3. Kubernetes Optimizer：在生产环境中我们一直以来都使用 External Flink Optimizer 来做 Iceberg 的 compaction 任务，这样的方式其实并不方便灵活的调整 optimizer 的资源。如果像 LocalOptimizer 一样弹性的管理 Pod，那可以一定程度上减少维护外部 Optimizer 的成本，也节省了一部分 JobManger 的资源。