Skip to main content
NetApp artificial intelligence solutions
简体中文版经机器翻译而成，仅供参考。如与英语版出现任何冲突，应以英语版为准。

8.配置参考

贡献者 nkarthik
PDF

Karthikeyan Nagalingam， NetApp

配置参考描述了控制数据摄取、数据准备、XCP 移动性、训练、检查点和存档的运行时参数。ONTAP NAS 存储桶主要用作可支持性和可移植性的预备数据和 XCP 暂存层。在实时生产使用中，当更符合工作负载和延迟要求时，也可以通过支持 RDMA 的直接访问路径使用相同的预备数据。

所有运行时行为均通过 `dag_run.conf`提供，使同一 DAG 能够服务于不同的源系统、转换引擎、存储目标、恢复策略和归档范围。仅配置所选工作流所需的参数组，将凭据保存在 Airflow Connections 或 secrets 后端中，并将示例用作非生产模板，而非凭据值。

[[8-1-ingestion-parameters]]
== 8.1 采集参数

使用这些设置来选择和配置上游摄取机制。 s3_direct 使用数据准备组中配置的 StorageGRID 原始数据对象；仅当选择了相应的摄取工具时，才需要 Airbyte 和 NiFi 设置。

参数 默认 必填项 描述 示例

ingestion_tool

s3_direct

否

选择摄取模式

"s3_direct"

airbyte_api_url

无

仅限 Airbyte

Airbyte API 端点

"http://airbyte:8000"

airbyte_connection_id

无

仅限 Airbyte

Airbyte 连接 ID

"connection-123"

airbyte_api_token

无

可选

Bearer 令牌

"<TOKEN>"

airbyte_timeout_secs

30

否

Airbyte 超时

60

nifi_api_url

无

NiFi 专用

NiFi API 端点

"http://nifi:8080"

nifi_process_group_id

无

NiFi 专用

NiFi 流程组

"abc123"

nifi_timeout_secs

30

否

NiFi 超时

60

[[8-2-data-preparation-parameters]]
== 8.2 数据准备参数

这些设置控制原始数据访问、准备好的数据输出、输入发现和转换行为。 s3_raw_* 参数指的是 StorageGRID，而代码兼容的 s3_formatted_* 参数标识写入已准备数据、运行标记数据和清单的 ONTAP NAS 存储桶。

原始 StorageGRID / 准备好的数据 ONTAP NAS 存储桶：

当 StorageGRID 与 ONTAP NAS 存储桶使用不同的 S3 兼容服务或访问策略时，请配置单独的端点和凭据。通用回退密钥仅在缺少更具体的原始数据或准备好的数据设置时才适用。

参数 默认 描述 示例

s3_raw_bucket

ai-raw-data

StorageGRID 原始输入存储桶

"bucket1"

s3_raw_prefix

example_ai_pipeline/raw

StorageGRID 原始数据前缀

"example_ai_pipeline/raw"

s3_raw_endpoint_url

后备链

StorageGRID S3 端点

"http://10.63.150.62:10444"

s3_raw_access_key_id

后备链

原始访问密钥

"<KEY>"

s3_raw_secret_access_key

后备链

原始密钥

"<SECRET>"

s3_raw_region

aws_region

原始区域

"us-east-1"

s3_formatted_bucket

ai-formatted-data

ONTAP NAS 准备好的数据输出桶

"prepnasbucket"

s3_formatted_prefix

example_ai_pipeline/formatted

ONTAP NAS 准备好的数据前缀

"example_ai_pipeline/formatted"

s3_formatted_endpoint_url

后备链

ONTAP NAS bucket S3 端点

"http://10.63.150.159"

s3_formatted_access_key_id

后备链

ONTAP NAS 存储桶访问密钥

"<KEY>"

s3_formatted_secret_access_key

后备链

ONTAP NAS 存储桶密钥

"<SECRET>"

一般回退： s3_endpoint_url， s3_access_key_id， s3_secret_access_key， s3_session_token， s3_verify， aws_region， aws_access_key_id， aws_secret_access_key， aws_session_token。

输入选择：

使用显式对象键设置来处理已知的 CSV 输入；否则，任务会在配置的 StorageGRID 原始前缀下发现合格的表格对象。

参数 默认 描述 示例

s3_tabular_object_keys

自动发现

CSV 密钥的显式列表

["raw/demo/a.csv"]

s3_tabular_object_key

无

单个显式CSV密钥

"raw/demo/a.csv"

转型：

选择 Python 路径进行轻量级本地准备，或选择 Spark 进行分布式准备。采样和种子可重复地应用于生成的训练、验证和推理拆分。

参数 默认 描述 示例

transformation_engine

python

python 或 spark

"spark"

spark_required

false

失败而非回退

true

table_format

none

none/delta/iceberg

"delta"

table_format_required

false

格式不可用时失败

true

sample_count

所有行

行限制（≤0 = 全部）

14400

seed

7

可重复的随机种子

11

Spark:

这些设置仅适用于 transformation_engine=spark。它们控制 Spark 进程放置、资源分配、时间限制和可选的 Delta Lake 或 Iceberg 运行时依赖项。

参数 默认 描述 示例

spark_submit_bin

spark-submit

Spark 二进制文件

"/opt/spark/bin/spark-submit"

spark_master

local[*]

Spark master

"local[*]"

spark_driver_memory

4g

驱动程序内存

"8g"

spark_executor_memory

4g

执行器内存

"8g"

spark_timeout_secs

900/7200

超时（0 = 禁用）

0

spark_delta_packages

Delta 3.2.0

Delta 运行时包

"io.delta:delta-spark_2.12:3.2.0"

spark_iceberg_packages

Iceberg 1.5.2

Iceberg 运行时软件包

"org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.5.2"

spark_iceberg_catalog

local

Iceberg 目录名称

"local"

[[8-3-xcp-and-training-destination-parameters]]
== 8.3 XCP 和 Training 目标参数

在 `enable_xcp=true`验证 ONTAP NAS NFS 源、调用 XCP 主机并选择活动训练层时使用此组。 `xcp_copy_destination`是控制开关：它选择下面的 ONTAP S3 特定或 LustreFS 特定设置，下游模型阶段使用相同的选定层。

参数 默认 必填项 描述 示例

enable_xcp

false

是，适用于 XCP

启用 XCP 分支

true

xcp_copy_destination

s3

否

s3 或 lustrefs

"lustrefs"

xcp_nfs_source_ip

10.63.150.159

XCP 预检

NFS 服务器 IP

"10.63.150.159"

xcp_nfs_source_path

/prepnasbucket/example_ai_pipeline

XCP 预检

NFS 导出路径

"/prepnasbucket/example_ai_pipeline"

ssh_user

root

否

XCP 主机 SSH 用户

"root"

ssh_host

10.63.150.178

否

XCP 主机地址

"10.63.150.178"

S3 目标（以下情况为必填项 xcp_copy_destination=s3）：

仅为 ONTAP S3 培训路径提供这些设置。配置文件或直接凭据允许 XCP 和模型工件同步逻辑访问所选的 ONTAP S3 存储桶和前缀。

参数 默认 描述 示例

xcp_s3_endpoint

默认

XCP S3 端点

"http://10.63.150.161"

xcp_s3_bucket

无

XCP 目标 Bucket

"trainingbucket"

xcp_s3_dest_prefix

example_ai_pipeline

目标前缀

"example_ai_pipeline"

xcp_s3_training_prefix

xcp_s3_dest_prefix

训练发现前缀

"example_ai_pipeline"

xcp_s3_profile

ontaps3

已命名凭据配置文件

"ontaps3"

xcp_s3_profiles

无

配置文件映射

{"ontaps3": {…​}}

xcp_s3_access_key_id

无

直接密钥回退

"<KEY>"

xcp_s3_secret_access_key

无

直接秘密回退

"<SECRET>"

xcp_s3_region

aws_region

XCP S3 区域

"us-east-1"

LustreFS 目标（当 `xcp_copy_destination=lustrefs`时为必填项）：

仅为 LustreFS 培训路径提供这些设置。源和目标必须在 XCP 主机上挂载并可访问；运行模型阶段的 Airflow worker 也必须可访问目标。

参数 默认 描述 示例

xcp_lustrefs_source_path

/prepnasbucket/example_ai_pipeline

XCP 复制源

"/prepnasbucket/example_ai_pipeline"

xcp_lustrefs_dest_path

/mnt/lustre/client

LustreFS 挂载目标

"/mnt/lustre/client"

xcp_lustrefs_training_subpath

xcp_s3_dest_prefix

挂载下的训练子目录

"example_ai_pipeline"

xcp_lustrefs_newid

data_prep 运行 stamp

XCP 作业标识符

"20260901_120000"

文本源行为：

这些选项控制在 XCP 移动性之后如何找到三个文本数据集。显式键覆盖发现；可以禁用本地回退，以要求文本输入来自所选 XCP 目标。

参数 默认 描述 示例

xcp_text_allow_local_fallback

true

允许本地文本回退

false

xcp_text_base_key

自动

显式基准文本位置

"…​/text_base.json"

xcp_text_finetune_key

自动

显式微调文本位置

"…​/text_finetune.json"

xcp_text_infer_key

自动

显式推理文本位置

"…​/text_infer.json"

[[8-4-model-training-parameters]]
== 8.4 模型训练参数

这些设置选择本地或 XCP 提供的源，限制训练量，并保留可重现的数据排序。启用 XCP 后，训练将读取所选的 ONTAP S3 或 LustreFS 层，并将基线工件发布回同一目标。

参数 默认 描述 示例

enable_xcp

false

选择XCP或本地源

true

xcp_copy_destination

s3

选择源层

"lustrefs"

sample_count

所有行

训练样本限制

14400

seed

7

可重现的随机播放

11

产生： regression_model.bin、 text_vectorizer.bin、 text_classifier.bin、 train_metrics.json。

[[8-5-fine-tuning-parameters]]
== 8.5 微调参数

微调从所选 XCP 目标实现基线文本工件，将增量学习应用于微调数据集，并重新发布经过调整的分类器及其指标。保持其 XCP 设置与模型训练一致，以维护工件溯源。

参数 默认 描述 示例

enable_xcp

false

启用制品物化/发布

true

xcp_copy_destination

s3

选择制品来源/目的地

"lustrefs"

消耗： text_vectorizer.bin， text_classifier.bin， text_finetune.json。产生： text_classifier_tuned.bin， fine_tune_metrics.json。

[[8-6-inference-parameters]]
== 8.6 推理参数

这些标志控制从所选训练层实现调优模型后的评分范围。当验证或业务需求需要超出默认推理拆分和推理文本输入的输出时，启用任一选项。

参数 默认 描述 示例

infer_all_splits

false

评分训练/验证/推理拆分

true

infer_all_text

false

得分基础/微调/推断文本

true

消耗： regression_model.bin， text_vectorizer.bin， text_classifier_tuned.bin。产生： tabular_predictions.csv， text_predictions.json。

[[8-7-checkpoint-parameters]]
== 8.7 检查点参数

检查点作用域为 model_training。这些设置确定是否创建训练完成记录、存储位置、如何处理上传失败，以及后续运行是否验证或重复使用有效的已完成训练结果。

参数 默认 描述 示例

checkpoint_enabled

true

启用检查点文件创建

true

checkpoint_store

local

local 或 formatted_s3

"formatted_s3"

checkpoint_upload_fail_mode

warn

warn 或 fail

"fail"

training_checkpoint_reuse_mode

off

off/verify_only/resume_if_exists

"resume_if_exists"

checkpoint_reuse_mode

off

向后兼容的别名

"verify_only"

[[8-8-manual-archive-parameters]]
== 8.8 手动归档参数

使用此组可启用 StorageGRID 存档，标识目标存储桶和凭据，并选择存档点。所选的 `manual_archive_stage`控制存档时间以及存档是否仅包含基线训练工件或完整的推理结果集。

参数 默认 必填项 描述 示例

manual_archive_enabled

false

否

启用存档

true

manual_archive_stage

inferencing

否

归档点和工件集： `model_training`用于核心训练工件，或 `inferencing`用于完整结果集

"model_training"

manual_archive_bucket

archivalbucket

启用时为是

归档分段

"archivalbucket"

manual_archive_prefix

ai-models

否

归档前缀

"ai-models"

manual_archive_endpoint

默认

仅限自定义

归档 S3 端点

"http://10.63.150.62:10444"

manual_archive_profile

sgdlocal

建议

凭据配置文件

"sgdlocal"

manual_archive_profiles

无

可选

配置文件映射

{"sgdlocal": {…​}}

manual_archive_include_xcp_inputs

false

否

在存档中包含 XCP 输入

true

manual_archive_cleanup_raw

false

否

上传后移除本地原始文件夹

true

别名： `fabricpool_archive_*`用于所有 `manual_archive_*`参数。

[[8-8-1-manual-archive-stage-decision]]
=== 8.8.1 手动存档阶段决策

The route_manual_archive_stage 分支任务每次 DAG 运行读取 `manual_archive_stage`一次。所选值确定何时运行存档以及将哪些工件上传到 StorageGRID 存档存储桶。存档对象存储在 `s3://<manual_archive_bucket>/<manual_archive_prefix>/<stage>/<run_stamp>/`下。

manual_archive_stage 值 归档时间 工件已存档 建议使用

model_training

此后立即 model_training；不等待微调或推理

regression_model.bin, text_vectorizer.bin, text_classifier.bin, train_metrics.json

保留可重现的基线模型，最大限度地减少存档量，或在后期阶段之前保留检查点

inferencing (默认)

在 fine_tuning 和 inferencing 完成后

所有 model_training`工件加上 `text_classifier_tuned.bin、 fine_tune_metrics.json、 tabular_predictions.csv`和 `text_predictions.json

保留模型运行的完整业务结果和预测证据

对于任一选择，设置 `manual_archive_include_xcp_inputs=true`以在存在时添加本地实例化的 XCP 训练输入。设置 `manual_archive_enabled=false`以跳过存档任务，而不更改训练和推理工作流程。

[[8-9-complete-sample-configuration]]
== 8.9 完整示例配置

此目录提供主体部署和测试路径的启动配置。每个示例都假设原始数据位于 StorageGRID，并且准备好的数据会写入 ONTAP NAS 存储桶中。将存储桶名称、端点地址、文件系统路径和配置文件名称替换为目标环境中的值。请勿将生产访问密钥或机密放在 `CONF_JSON`中；请通过 Airflow Connections、Variables 或 secrets backend 进行配置。

[[8-9-1-xcp-to-lustrefs-with-spark-and-delta-lake]]
=== 8.9.1 XCP 到 LustreFS（使用 Spark 和 Delta Lake）

使用此全吞吐量示例进行分布式准备和高性能 LustreFS 模型 I/O。它处理 `s3_raw_prefix`下所有符合条件的表格文件，在准备期间写入 Delta 表，将训练检查点持久化到 ONTAP NAS 准备好的数据桶中，并存档完整的推断结果集。

{
  "ingestion_tool": "s3_direct",
  "s3_raw_bucket": "bucket1",
  "s3_raw_prefix": "example_ai_pipeline/raw",
  "s3_formatted_bucket": "prepnasbucket",
  "s3_formatted_prefix": "example_ai_pipeline/formatted",

  "enable_xcp": true,
  "xcp_copy_destination": "lustrefs",
  "xcp_nfs_source_ip": "10.63.150.159",
  "xcp_nfs_source_path": "/prepnasbucket/example_ai_pipeline",
  "xcp_lustrefs_source_path": "/prepnasbucket/example_ai_pipeline",
  "xcp_lustrefs_dest_path": "/mnt/lustre/client",
  "xcp_lustrefs_training_subpath": "example_ai_pipeline",
  "xcp_text_allow_local_fallback": false,

  "transformation_engine": "spark",
  "spark_required": true,
  "table_format": "delta",
  "table_format_required": true,
  "spark_driver_memory": "8g",
  "spark_executor_memory": "8g",
  "spark_timeout_secs": 0,
  "sample_count": 14400,
  "seed": 11,

  "infer_all_splits": true,
  "infer_all_text": true,

  "checkpoint_enabled": true,
  "checkpoint_store": "formatted_s3",
  "training_checkpoint_reuse_mode": "off",

  "manual_archive_enabled": true,
  "manual_archive_stage": "inferencing",
  "manual_archive_bucket": "archivalbucket",
  "manual_archive_prefix": "ai-models",
  "manual_archive_profile": "sgdlocal",
  "manual_archive_cleanup_raw": true
}

[[8-9-2-no-xcp-with-python-preparation]]
=== 8.9.2 没有使用 Python 准备的 XCP

将此基线配置用于轻量级功能运行。准备阶段使用本地 Python 引擎，下游阶段使用管道的本地准备数据和制品路径；XCP、Spark、表格格式、检查点重用和存档已禁用。

{
    "ingestion_tool": "s3_direct",
    "s3_raw_bucket": "bucket1",
    "s3_raw_prefix": "example_ai_pipeline/raw",
    "s3_formatted_bucket": "prepnasbucket",
    "s3_formatted_prefix": "example_ai_pipeline/formatted",
    "enable_xcp": false,
    "transformation_engine": "python",
    "table_format": "none",
    "sample_count": 200,
    "seed": 11,
    "checkpoint_enabled": false,
    "manual_archive_enabled": false
}

[[8-9-3-python-preparation-with-two-explicit-input-files]]
=== 8.9.3 使用两个显式输入文件准备 Python

针对两个已知的 StorageGRID CSV 对象验证管道时，请使用此重点测试。 `s3_tabular_object_keys`禁用自动表格文件发现；文本输入仍使用原始前缀下的预期位置。

{
    "ingestion_tool": "s3_direct",
    "s3_raw_bucket": "bucket1",
    "s3_raw_prefix": "example_ai_pipeline/raw",
    "s3_tabular_object_keys": [
        "example_ai_pipeline/raw/tabular_part_01.csv",
        "example_ai_pipeline/raw/tabular_part_02.csv"
    ],
    "s3_formatted_bucket": "prepnasbucket",
    "s3_formatted_prefix": "example_ai_pipeline/formatted",
    "enable_xcp": false,
    "transformation_engine": "python",
    "table_format": "none",
    "sample_count": 200,
    "seed": 11,
    "manual_archive_enabled": false
}

[[8-9-4-python-preparation-with-automatic-all-file-discovery]]
=== 8.9.4 使用自动全文件发现的 Python 准备

省略 `s3_tabular_object_keys`和 `s3_tabular_object_key`以处理 StorageGRID 原始前缀下的每个符合条件的表格对象。此示例适用于使用 Python 准备引擎的全文件功能或规模测试。

{
    "s3_raw_bucket": "bucket1",
    "s3_raw_prefix": "example_ai_pipeline/raw",
    "s3_formatted_bucket": "prepnasbucket",
    "s3_formatted_prefix": "example_ai_pipeline/formatted",
    "enable_xcp": false,
    "transformation_engine": "python",
    "table_format": "none",
    "sample_count": 0,
    "seed": 11,
    "manual_archive_enabled": false
}

sample_count: 0 表示数据准备使用所有可用行。为有界测试运行设置正值。

[[8-9-5-xcp-to-ontap-s3-with-python-preparation]]
=== 8.9.5 XCP 到 ONTAP S3 与 Python 的准备

当必须将准备好的数据从 ONTAP NAS NFS 导出复制到 ONTAP S3 训练存储桶时，请使用此配置。模型训练、微调和推理通过相同的 ONTAP S3 目标实现并发布工件。

{
    "s3_raw_bucket": "bucket1",
    "s3_raw_prefix": "example_ai_pipeline/raw",
    "s3_formatted_bucket": "prepnasbucket",
    "s3_formatted_prefix": "example_ai_pipeline/formatted",
    "enable_xcp": true,
    "xcp_copy_destination": "s3",
    "xcp_nfs_source_ip": "10.63.150.159",
    "xcp_nfs_source_path": "/prepnasbucket/example_ai_pipeline",
    "xcp_s3_endpoint": "https://ontap-s3.example.com",
    "xcp_s3_bucket": "trainingbucket",
    "xcp_s3_dest_prefix": "example_ai_pipeline",
    "xcp_s3_training_prefix": "example_ai_pipeline",
    "xcp_s3_profile": "ontaps3",
    "transformation_engine": "python",
    "table_format": "none",
    "sample_count": 14400,
    "seed": 11,
    "checkpoint_enabled": true,
    "checkpoint_store": "formatted_s3",
    "manual_archive_enabled": true,
    "manual_archive_stage": "model_training",
    "manual_archive_bucket": "archivalbucket",
    "manual_archive_profile": "storagegrid-archive"
}

[[8-9-6-xcp-to-ontap-s3-with-spark-and-iceberg]]
=== 8.9.6 使用 Spark 和 Iceberg 的 XCP 到 ONTAP S3

使用此配置进行分布式准备，以 Iceberg 表输出和 ONTAP S3 作为活动训练目标。当 Spark 或 Iceberg 不可用时，如果运行必须失败而非回退，请将 spark_required`和 `table_format_required`设置为 `true。

{
    "s3_raw_bucket": "bucket1",
    "s3_raw_prefix": "example_ai_pipeline/raw",
    "s3_formatted_bucket": "prepnasbucket",
    "s3_formatted_prefix": "example_ai_pipeline/formatted",
    "enable_xcp": true,
    "xcp_copy_destination": "s3",
    "xcp_nfs_source_ip": "10.63.150.159",
    "xcp_nfs_source_path": "/prepnasbucket/example_ai_pipeline",
    "xcp_s3_bucket": "trainingbucket",
    "xcp_s3_dest_prefix": "example_ai_pipeline",
    "xcp_s3_profile": "ontaps3",
    "transformation_engine": "spark",
    "spark_required": true,
    "table_format": "iceberg",
    "table_format_required": true,
    "spark_driver_memory": "8g",
    "spark_executor_memory": "8g",
    "spark_timeout_secs": 0,
    "sample_count": 0,
    "seed": 11,
    "manual_archive_enabled": true,
    "manual_archive_stage": "inferencing",
    "manual_archive_bucket": "archivalbucket",
    "manual_archive_profile": "storagegrid-archive"
}

[[8-9-7-scenario-selection-summary]]
=== 8.9.7 场景选择摘要

场景 XCP 目标 制备引擎 表格格式 输入范围 归档阶段

8.9.1

LustreFS

火花

Delta

所有已发现文件

inferencing

8.9.2

无

Python

无

所有已发现文件

已禁用

8.9.3

无

Python

无

两个显式文件

已禁用

8.9.4

无

Python

无

所有已发现文件

已禁用

8.9.5

ONTAP S3

Python

无

所有已发现文件

model_training

8.9.6

ONTAP S3

火花

Iceberg

所有已发现文件

inferencing

[[8-9-8-execution-walkthroughs]]
=== 8.9.8 执行演练

从 Airflow 工作区运行示例。触发器脚本生成唯一 `manual__<UTC timestamp>`的运行 ID，等待完成，并返回失败或超时运行的非零退出代码。以下完整配置使用与操作示例相同的 `CONF_JSON="$(python3 - <<'PY' …​)"`模式。在运行之前，在密钥后端配置引用的 Airflow 托管连接、XCP 配置文件和归档配置文件，并将 `CONF_JSON`限制为非密钥存储桶、端点、路径和配置文件标识符。

有效的表格式名称为 iceberg。请勿使用 icerberg，此值不受支持。

所有文件、Spark、Iceberg、XCP 到 ONTAP S3、检查点到 prepared-data S3、推断后归档：

CONF_JSON="$(python3 - <<'PY'
import json

print(json.dumps({
    "ingestion_tool": "s3_direct",
    "enable_xcp": True,
    "xcp_copy_destination": "s3",
    "xcp_text_allow_local_fallback": True,

    "s3_raw_bucket": "bucket1",
    "s3_raw_prefix": "example_ai_pipeline/raw",
    "s3_raw_endpoint_url": "https://storagegrid.example.com",
    "s3_raw_region": "us-east-1",

    "s3_formatted_bucket": "prepnasbucket",
    "s3_formatted_prefix": "example_ai_pipeline/formatted",
    "s3_formatted_endpoint_url": "https://ontap-nas.example.com",
    "s3_formatted_region": "us-east-1",

    "xcp_nfs_source_ip": "10.63.150.159",
    "xcp_nfs_source_path": "/prepnasbucket/example_ai_pipeline",
    "xcp_s3_endpoint": "https://ontap-s3.example.com",
    "xcp_s3_bucket": "trainingbucket",
    "xcp_s3_dest_prefix": "example_ai_pipeline",
    "xcp_s3_training_prefix": "example_ai_pipeline",
    "xcp_s3_profile": "ontaps3",

    "transformation_engine": "spark",
    "spark_required": True,
    "table_format": "iceberg",
    "table_format_required": True,
    "sample_count": 0,
    "seed": 11,
    "spark_driver_memory": "8g",
    "spark_executor_memory": "8g",
    "spark_timeout_secs": 0,
    "infer_all_splits": True,
    "infer_all_text": True,

    "checkpoint_enabled": True,
    "checkpoint_store": "formatted_s3",
    "manual_archive_enabled": True,
    "manual_archive_stage": "inferencing",
    "manual_archive_bucket": "archivalbucket",
    "manual_archive_prefix": "ai-models",
    "manual_archive_profile": "sgdlocal",
    "manual_archive_endpoint": "https://storagegrid.example.com",
    "manual_archive_cleanup_raw": True,
    "manual_archive_include_xcp_inputs": True,
}, separators=(",", ":")))
PY
)"
CONF_JSON="$CONF_JSON" AIRFLOW_BIN=./.venv/bin/airflow INGESTION_TOOL=s3_direct \
  ./trigger_and_wait_ai_pipeline_sklearn.sh

所有文件、Spark、Iceberg、XCP 到 LustreFS、检查点到准备好的数据 S3、推断后归档：

CONF_JSON="$(python3 - <<'PY'
import json

print(json.dumps({
    "ingestion_tool": "s3_direct",
    "enable_xcp": True,
    "xcp_copy_destination": "lustrefs",
    "ssh_user": "root",
    "ssh_host": "10.63.150.178",
    "xcp_text_allow_local_fallback": True,

    "s3_raw_bucket": "bucket1",
    "s3_raw_prefix": "example_ai_pipeline/raw",
    "s3_raw_endpoint_url": "https://storagegrid.example.com",
    "s3_raw_region": "us-east-1",

    "s3_formatted_bucket": "prepnasbucket",
    "s3_formatted_prefix": "example_ai_pipeline/formatted",
    "s3_formatted_endpoint_url": "https://ontap-nas.example.com",
    "s3_formatted_region": "us-east-1",

    "xcp_nfs_source_ip": "10.63.150.159",
    "xcp_nfs_source_path": "/prepnasbucket/example_ai_pipeline",
    "xcp_lustrefs_source_path": "/prepnasbucket/example_ai_pipeline",
    "xcp_lustrefs_dest_path": "/mnt/lustre/client",
    "xcp_lustrefs_training_subpath": "example_ai_pipeline",

    "transformation_engine": "spark",
    "spark_required": True,
    "table_format": "iceberg",
    "table_format_required": True,
    "sample_count": 0,
    "seed": 11,
    "spark_driver_memory": "8g",
    "spark_executor_memory": "8g",
    "spark_timeout_secs": 0,
    "infer_all_splits": True,
    "infer_all_text": True,

    "checkpoint_enabled": True,
    "checkpoint_store": "formatted_s3",
    "manual_archive_enabled": True,
    "manual_archive_stage": "inferencing",
    "manual_archive_bucket": "archivalbucket",
    "manual_archive_prefix": "ai-models",
    "manual_archive_profile": "sgdlocal",
    "manual_archive_endpoint": "https://storagegrid.example.com",
    "manual_archive_cleanup_raw": True,
    "manual_archive_include_xcp_inputs": True,
}, separators=(",", ":")))
PY
)"
AIRFLOW_BIN=./.venv/bin/airflow INGESTION_TOOL=s3_direct \
  ./trigger_and_wait_ai_pipeline_sklearn.sh

对于 ONTAP S3 模式，在运行之前，仅将非密钥 xcp_s3_profile`名称传入 `CONF_JSON，并在 XCP 主机上或通过 Airflow 管理的密钥存储定义相应的凭据。对于 LustreFS 模式，由于训练路由不解析 ONTAP S3 凭据，因此会故意省略 xcp_s3_profiles`映射。在这两个示例中，通过 Airflow 管理的密钥存储而不是内联 JSON 来定义 StorageGRID 原始数据、ONTAP NAS 准备好的数据和存档凭据。省略 `s3_tabular_object_keys`并将 `sample_count`设置为 `0，以请求自动发现所有符合条件的表格文件并使用所有可用行。

Python，无 XCP，双文件测试：

CONF_JSON='{"s3_raw_bucket":"bucket1","s3_raw_prefix":"example_ai_pipeline/raw","s3_tabular_object_keys":["example_ai_pipeline/raw/tabular_part_01.csv","example_ai_pipeline/raw/tabular_part_02.csv"],"s3_formatted_bucket":"prepnasbucket","s3_formatted_prefix":"example_ai_pipeline/formatted","enable_xcp":false,"transformation_engine":"python","table_format":"none","sample_count":200,"manual_archive_enabled":false}' \
./trigger_and_wait_ai_pipeline_sklearn.sh

Spark、Delta Lake、XCP to LustreFS：

CONF_JSON='{"enable_xcp":true,"xcp_copy_destination":"lustrefs","xcp_nfs_source_ip":"10.63.150.159","xcp_nfs_source_path":"/prepnasbucket/example_ai_pipeline","xcp_lustrefs_source_path":"/prepnasbucket/example_ai_pipeline","xcp_lustrefs_dest_path":"/mnt/lustre/client","xcp_lustrefs_training_subpath":"example_ai_pipeline","transformation_engine":"spark","spark_required":true,"table_format":"delta","table_format_required":true,"sample_count":14400,"manual_archive_enabled":true,"manual_archive_stage":"inferencing"}' \
./trigger_and_wait_ai_pipeline_sklearn.sh

Spark、Iceberg、XCP to ONTAP S3：

CONF_JSON='{"enable_xcp":true,"xcp_copy_destination":"s3","xcp_nfs_source_ip":"10.63.150.159","xcp_nfs_source_path":"/prepnasbucket/example_ai_pipeline","xcp_s3_bucket":"trainingbucket","xcp_s3_dest_prefix":"example_ai_pipeline","xcp_s3_profile":"ontaps3","transformation_engine":"spark","spark_required":true,"table_format":"iceberg","table_format_required":true,"sample_count":0,"manual_archive_enabled":true,"manual_archive_stage":"inferencing"}' \
./trigger_and_wait_ai_pipeline_sklearn.sh

在 XCP 运行之前，从 XCP 主机验证配置的 ONTAP NAS NFS 路径和目标可访问性。成功运行后，请查看 [data_prep] effective_config、 [xcp_copy] effective_config、 [model_training] effective_config 和 `[manual_archive]`的任务日志，以确认所选引擎、输入范围、目标和存档阶段。