Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异，请参阅 中国的 Amazon Web Services 服务入门 (PDF)。

本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

自动优化

Auto-Optimize 是一项自动优化矢量索引的服务，使用户无需数周的专家手动调整即可平衡搜索质量、速度和成本。它根据用户定义的延迟和召回要求评估索引配置，并生成优化建议，因此只需要最少的专业知识。建议通常在 30-60 分钟内送达。

传统的向量索引配置需要大量的专业知识和实验才能实现最佳性能。诸如ef_construction（控制索引构建质量）（决定图形连接数）m（控制HNSW搜索）和量化方法ef_search（二进制量化（32x、16x、8x）、标量量化（4x））之类的参数会显著影响搜索准确性和资源利用率。Auto-Optimize 使用超参数优化算法，在您定义的延迟和召回要求范围内发现最适合您的数据集的索引配置。

优势

自动优化 OpenSearch 具有以下好处：

自动优化与其他 OpenSearch 功能配合使用，例如用于向量索引的 GPU 加速为矢量搜索工作负载提供全面的性能优化。

工作原理

自动优化通过基于作业的架构运行，该架构可分析您的矢量数据并提供优化建议。关键点：

先决条件

自动优化的用例

在以下情况下，自动优化特别有用：

初始配置优化

首次部署矢量搜索应用程序时，确定最佳 HNSW 参数通常需要大量的测试和领域专业知识。自动优化通过分析您的数据和工作负载特征来推荐生产就绪配置，从而消除了这一 trial-and-error过程。

此用例非常适合刚接触矢量搜索的团队或需要快速建立基准配置的从其他矢量数据库平台迁移的团队。

缩放优化

当您的矢量数据集从数千个向量增加到数百万个向量时，最初效果良好的参数可能会变得不理想。自动优化建议进行调整以保持大规模性能。

降低成本

向量索引会消耗大量的计算和存储资源，对于高维嵌入尤其如此。Auto-Optimize 通过寻找更有效的参数配置来发现降低成本的机会，这些配置既能保持所需的性能水平，又能使用更少的资源。

例如，自动优化可能会发现您的当前m值（图形连接性）高于精度要求所需的值，从而使您能够在不影响搜索质量的情况下缩短索引时间和存储空间。

性能故障排除

当在向量搜索操作中遇到查询性能缓慢或延迟高时，自动优化可以分析您的数据集并确定更优的配置。该服务提供解决性能瓶颈的具体建议，例如调整图形连接或搜索参数。

限制

账单和成本

Auto-Optimize 使用按作业定价模型，即无论数据集大小和优化配置如何，您都要为每项成功的优化作业付费。您不会因为失败或取消的任务而被收费。此外，自动优化在与托管或无服务器 OpenSearch 集群不同的基础架构上运行，因此它不会影响先前存在的集群的资源利用率。

定价模式

自动优化费用与标准 OpenSearch 无服务器或 OpenSearch 托管域计算和存储成本分开计费。

有关定价信息，请参阅 Amazon OpenSearch 服务定价。

将 JSONL 转换为实木复合地板

如果您的数据采用 JSONL 格式，则可以使用以下 Python 脚本将其转换为 Parquet 格式以用于自动优化：

#!/usr/bin/env python3
import json
import pyarrow as pa
import pyarrow.parquet as pq
from pathlib import Path
from typing import Any, Dict, List


def load_json_any(path: Path) -> List[Dict[str, Any]]:
    """
    Load JSON that can be:
      - a list of objects
      - a single object
      - JSON Lines (one object per line)
    Returns list[dict].
    """
    text = path.read_text().strip()

    # Try full JSON file
    try:
        obj = json.loads(text)
        if isinstance(obj, list):
            return obj
        if isinstance(obj, dict):
            return [obj]
    except json.JSONDecodeError:
        pass

    # Fallback → JSON Lines
    records = []
    for i, line in enumerate(text.splitlines(), start=1):
        line = line.strip()
        if not line:
            continue
        try:
            rec = json.loads(line)
        except json.JSONDecodeError as e:
            raise ValueError(f"Invalid JSON on line {i}: {e}")
        if not isinstance(rec, dict):
            raise ValueError(f"Line {i} must contain a JSON object")
        records.append(rec)

    return records


def json_to_parquet(json_path: str, parquet_path: str, compression: str = "snappy"):
    """Convert ANY JSON to Parquet (schema inferred)."""
    records = load_json_any(Path(json_path))
    table = pa.Table.from_pylist(records)
    pq.write_table(table, parquet_path, compression=compression)
    print(f"✔ Wrote {len(records)} rows to {parquet_path}")


def print_parquet_rows(parquet_path: str, limit: int = 5):
    """Print first N rows from the parquet file."""
    table = pq.read_table(parquet_path)
    df = table.to_pandas()

    print(f"\n=== Showing first {min(limit, len(df))} rows from {parquet_path} ===")
    print(df.head(limit).to_string())
    print("===========================================================\n")


if __name__ == "__main__":
    INPUT_JSON = "movies_10k.json"
    OUTPUT_PARQUET = "movies.parquet"

    # Convert JSON → Parquet
    json_to_parquet(INPUT_JSON, OUTPUT_PARQUET)

    # Print some rows from Parquet
    print_parquet_rows(OUTPUT_PARQUET, limit=3)

相关功能

自动优化可与其他 Amazon S OpenSearch ervice 功能配合使用，帮助您构建和优化矢量搜索应用程序：