跳到主要内容
版本：Next

顶层配置

一份 ArkFlow 配置描述整个引擎:日志、健康检查 / 控制平面服务器、要运行的流列表,以及由统一内核(unified kernel)执行的可选流式 jobs。文件格式由扩展名决定——.yaml/.yml、.json 或 .toml 均可接受。

logging:
  level: info

health_check:
  enabled: true
  address: "127.0.0.1:8080"

streams:
  - id: orders
    input:
      type: memory        # any registered input — see the component pages
    pipeline:
      processors: []      # optional processors
    output:
      type: drop          # any registered output

jobs: []      # optional declarative streaming jobs, see "job" below

顶层字段​

字段类型必填默认值描述
streamsarray<stream>是*—要运行的流。
jobsarray<job>否[]由统一内核运行的声明式流作业(DAG + 时间 + 状态 + 检查点)。
loggingobject否见下文日志配置。
health_checkobject否见下文健康检查与控制平面服务器。

* streams 与 jobs 都默认为空列表;仅包含作业的配置是合法的(声明 streams: [] 或直接省略)。

logging​

字段类型必填默认值描述
levelstring否info日志级别:debug、info、warn、error。
file_pathstring否—把日志写入该文件而不是 stdout。
formatstring否plain日志格式:plain 或 json。

health_check​

运行一个带有 /health、/readiness 与 /liveness 端点的 HTTP 服务器(对 Kubernetes 有用)。当设置了 hub_urls 时,同一服务器还承载可选的控制平面 API 与 Hub Agent(参见控制平面)。

字段类型必填默认值描述
enabledboolean否true启动健康检查 / 控制平面服务器。
addressstring否127.0.0.1:8080监听地址。
health_pathstring否/health整体健康端点路径。
readiness_pathstring否/readiness就绪端点路径。
liveness_pathstring否/liveness存活端点路径。
api_prefixstring否/api/v1带版本的控制平面 API 前缀。
api_tokenstring否—保护带版本控制平面 API 的可选 Bearer 令牌。设置后,/api/v1 下的所有端点(含读端点)在没有有效 Authorization: Bearer 请求头时一律返回 401。
insecure_localboolean否false显式允许独立控制 API 在不设 api_token 的情况下绑定非回环 address。默认拒绝该组合启动;设置后进程会启动并输出 error 级警告。切勿在共享网络中使用。
cors_originsarray<string>否[]允许调用控制 API 的浏览器来源。为空则拒绝跨域调用。
hub_urlsarray<string>否[]计算节点 Agent 模式使用的 Hub 地址列表,按序作为故障转移候选(参见 Agent 多 Hub 故障转移)。空列表即为独立(standalone)模式。破坏性变更: 取代原先单字符串的 hub_url 字段;仍声明 hub_url 的配置会在校验时报错并附迁移指引——把 hub_url: "http://hub:8080" 改写为 hub_urls: ["http://hub:8080"]。
node_idstring否—本进程向其 Hub 上报的稳定身份。
node_tokenstring否—共享的节点注册凭据。绝不会包含在报告中。
agent_lease_ttl_msinteger否15000计算节点向其 Hub 通告的租约时长(毫秒)。
agent_session_ttl_msinteger否3600000Hub 签发的 Agent 会话凭据的硬性生命周期(毫秒);到期后 Agent 会透明地重新注册。
data_portinteger否—用于跨节点 shuffle 的数据面监听端口。缺省时节点不运行网络数据面,也绝不会通告 network_shuffle 能力。参与 split 放置的每个节点都要把它与 data_host 一起设置。
data_hoststring否—对等端用于访问本节点数据面的可路由主机(例如局域网 IP)。split 放置要求它与 data_port 一同设置。
observabilityobject否见下文进程级 Prometheus 指标与健康探针(参见可观测性)。

health_check.observability​

导出进程级可观测性端点。即使 health_check.enabled 为 false,它们依然可用——纯数据面部署(无控制平面 API、无 Hub)仍会暴露指标与探针。监听器默认绑定环回地址;生产环境请设置显式地址,或依赖主机/防火墙策略。这些端点在设计上不做认证——不要把该监听器暴露给不受信任的网络。启用控制平面服务器时,由其路由器提供相同端点,不会启动第二个监听器。

字段类型必填默认值描述
enabledboolean否true启动可观测性监听器(除非控制平面服务器已在提供这些端点)。
addressstring否127.0.0.1:8081监听地址。
metrics_pathstring否/metricsPrometheus 暴露端点(格式 0.0.4)。
ready_pathstring否/ready就绪探针:引擎完成启动所配置的流与作业后即成功。
live_pathstring否/live存活探针:进程运行期间即成功。

独立 Hub 的启动安全​

独立的 arkflow-server 控制平面默认 fail-closed。在把 Hub 绑定到环回地址之外之前,请将 ARKFLOW_HUB_STORAGE 设为持久化的 SQLite 路径,为操作员 API 设置 ARKFLOW_OPERATOR_TOKEN,为 Agent 注册设置 ARKFLOW_NODE_TOKEN。仅限明确的本地开发场景:在保持 ARKFLOW_HUB_ADDRESS 为环回地址的同时设置 ARKFLOW_HUB_INSECURE_LOCAL=1,这才允许易失状态与省略凭据。Hub 在绑定监听器之前会先恢复持久化状态,因此恢复错误会让服务器不可用,而不是提供一份残缺视图。

独立的 arkflow 引擎对其本地控制 API 施加同样的边界:health_check.address 为非回环地址且未设置 health_check.api_token 时拒绝启动。请设置 token,或用 health_check.insecure_local: true 显式确认(引擎随后启动并输出 error 级警告)。配置了 token 后,/api/v1 下的所有端点——含读端点——都要求该 Bearer 令牌;健康探针与顶层 /metrics 保持开放,供探针与抓取器使用。

备注

启用控制平面服务器时,/ready 与 /live 也会挂载在服务器地址上,与旧的 /health、/readiness、/liveness 端点并列(后者保持原有语义)。

Secret 引用​

敏感值(密码、令牌、密钥材料)不必明文写进配置文件。任何字符串值都可以内嵌引用,在配置物化为 EngineConfig 时解析一次——文件配置(--config)、--validate 以及经控制平面下发的配置均是如此。存储/下发的配置内容保持引用原文;只有运行中的进程在内存里持有解析出的值。

语法含义
${env:VAR}环境变量 VAR 的值;未设置则报错。
${env:VAR:-default}VAR 的值;未设置或为空时使用 default(${env:VAR:-} 允许显式空值)。
${file:/path}读取 /path 文件内容并剥离尾部换行(例如挂载的 Kubernetes Secret)。
${secret:NAME}环境变量 ARKFLOW_SECRET_<NAME> 的值(名字逐字映射)——为凭据提供独立命名空间。支持 :- 默认值。在 Hub 分发部署中,Hub 会在 rollout 分发时预解析这些引用,Agent 无需持有密钥环境。
$${字面 ${ 的转义。

规则与保证:

  • 引用可以出现在字符串值的任意位置,包括嵌套 map/array(host=${env:HOST};port=${env:PORT} 可行)。
  • 未知 scheme 的 ${...} 原样保留(向前兼容);键名与非字符串值绝不会被改动。
  • 解析结果不会被重扫:密钥值里若含 ${...} 保持字面(防注入)。
  • 解析错误只会指明配置路径与引用本身——绝不会包含密钥值。
logging:
  level: info

health_check:
  api_token: "${env:ARKFLOW_API_TOKEN:-}"

streams: []

若 ARKFLOW_API_TOKEN 未设置,进程会在启动时失败,错误形如 Failed to resolve secret reference at health_check.api_token: environment variable 'ARKFLOW_API_TOKEN' is not set (reference: ${env:ARKFLOW_API_TOKEN})。

备注

引用在所有配置物化处都是严格解析的。在 Hub–Agent 部署中,引用节点本地密钥的配置必须在密钥可解析的位置校验/应用(独立单进程部署不受影响)。中心化密钥存储将在后续 Hub 版本中提供。

stream​

streams 中的每个条目都是一条独立的处理管道。流字段的深入文档见组件一节;其结构为:

字段类型必填默认值描述
idstring否stream-<index>稳定的流标识符(必须唯一;用于 WAL 身份与 Hub 上报)。
inputobject是—输入组件(数据源)。
pipelineobject是—处理器管道。
outputobject是—输出组件(sink)。
error_outputobject否—接收处理器处理失败批次的输出。
bufferobject否—输入与处理器之间的缓冲 / 窗口策略。
durabilityobject否—流级 WAL 持久化(参见投递语义)。
stateobject否—旧式窗口的状态契约。不可恢复的窗口请声明 durability: ephemeral;需要持久状态则使用带检查点(checkpoint)的 jobs 条目。
temporaryarray<object>否—用于 join 的临时存储表。

pipeline​

字段类型必填默认值描述
thread_numinteger否1处理器工作任务数量。
processorsarray<object>是—处理器组件的有序列表。

job​

jobs 中的每个条目都是一个声明式流作业:带有显式事件时间、状态、检查点与恢复设置的算子 DAG。作业在本地与流一样通过统一内核运行;Hub 向计算节点分发的也是同样的作业结构(参见分布式作业)。

jobs:
  - id: local-job
    version: 1
    parallelism: 1
    max_parallelism: 128
    operators:
      - { id: source, kind: source }
      - { id: sink, kind: sink }
    edges:
      - { id: e1, from: source, to: sink, partitioned: true }
    sources:
      - operator_id: source
        input_type: generate
        config: { type: generate, context: '{"value": 1}', interval: 1s, batch_size: 10 }
        time:
          mode: processing_time
    sinks:
      - operator_id: sink
        output_type: stdout
    recovery: latest_checkpoint
字段类型必填默认值描述
idstring是—稳定的作业标识符;在 jobs 中必须唯一。
versioninteger是—作业版本;恢复时的状态格式兼容性依据它评估。
parallelisminteger否1默认任务并行度。
max_parallelisminteger否128用于键组(key-group)分区的上限。
operatorsarray<object>是—DAG 节点:id、kind(source、map、filter、aggregate、window、sink、udf);join 已保留,但在存在分布式多输入运行时之前会被拒绝。
edgesarray<object>否[]DAG 边:id、from、to、partitioned(按键组路由,而非同子任务)。
sourcesarray<object>否[]把组件输入附加到 source 算子:operator_id、input_type、config、time。
sinksarray<object>否[]把组件输出附加到 sink 算子:operator_id、output_type、config。
stateobject否—backend(例如 embedded_kv)、durability(默认 durable,或显式 ephemeral)、可选的稳定 root(或 ARKFLOW_STATE_ROOT)、namespace、ttl_ms、format_version、max_pending_transactions(正数;默认 4096;当窗口看到非常高的每窗口键基数时应调大,因为每个打开的窗口组或未确认的输出都会占用一个事务)、可选的正数 max_bytes 活动状态预算。有状态算子必需;持久状态还要求 checkpoint。
checkpointobject否—interval_ms、retention、object_store_uri(例如 file://... 或 s3://...)。
recoverystring否latest_checkpointlatest_checkpoint、latest_savepoint 或 fail。

time(source 的事件时间声明)​

字段类型必填默认值描述
modestring是—event_time 或 processing_time。
timestamp_fieldstring否—当 mode: event_time 时作为事件时间戳读取的字段。
watermarkobject否—strategy(bounded_out_of_orderness(默认)或 monotonous)、out_of_orderness_ms、idle_timeout_ms。
allowed_lateness_msinteger否0超过水印(Watermark)多久的迟到事件仍被接受。
late_event_policystring否drop迟到事件的处理方式:drop、route 或 update。
late_event_routestring否—当策略为 route 时接收被路由迟到事件的算子。
备注

中间算子类别(map、filter、aggregate、window、join、udf)目前主要由流式 SQL 编译器与控制台 DAG 编排器生成。部署前务必运行 --validate:它执行与启动时相同的深度构建检查,并显式拒绝不受支持的算子或状态后端。./target/release/arkflow schema 会输出权威的 JSON Schema(包含 jobs 字段),供编辑器补全使用。

运行前先校验​

务必先校验配置:

./target/release/arkflow --config config.yaml --validate

或者输出完整的 JSON Schema 并让编辑器指向它,以获得字段级补全:

./target/release/arkflow schema > arkflow.schema.json

文档附带一份预生成的 schema:/config-schema.json;编辑器配置参见 IDE 自动补全。