本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

自動最佳化

自動最佳化是一項服務，可自動化向量索引最佳化，讓使用者在搜尋品質、速度和成本之間取得平衡，而不需要數週的手動專家調校。它會根據使用者定義的延遲和召回需求來評估索引組態，並產生最佳化建議，因此需要最少的專業知識。建議通常會在 30-60 分鐘內交付。

傳統向量索引組態需要豐富的專業知識和實驗，才能達到最佳效能。ef_construction （控制索引建置品質）、 m （決定圖形連線數）、 ef_search （控制 HNSW 搜尋） 和量化方法 （二進位量化 (32x、16x、8x)、純量量化 (4x)) 等參數會大幅影響搜尋準確性和資源使用率。自動最佳化使用超參數最佳化演算法，在定義的延遲和召回需求內探索專屬於資料集的索引組態。

優勢

OpenSearch 的自動最佳化提供下列優點：

自動最佳化可與其他 OpenSearch 功能搭配使用，例如 向量索引的 GPU 加速，為向量搜尋工作負載提供全面的效能最佳化。

運作方式

自動最佳化透過以任務為基礎的架構運作，可分析向量資料並提供最佳化建議。重點：

先決條件

自動最佳化的使用案例

在下列案例中，自動最佳化特別有價值：

初始組態最佳化

第一次部署向量搜尋應用程式時，判斷最佳 HNSW 參數通常需要廣泛的測試和領域專業知識。自動最佳化可分析您的資料和工作負載特性，以建議生產就緒組態，藉此消除此trial-and-error程序。

此使用案例非常適合初次搜尋向量的團隊，或從需要快速建立基準組態的其他向量資料庫平台遷移的團隊。

擴展最佳化

隨著向量資料集從數千個向量成長到數百萬個向量，一開始運作良好的參數可能會變得不理想。自動最佳化建議調整，以大規模維持效能。

降低成本

向量索引可能會耗用大量的運算和儲存資源，特別是在高維度內嵌的情況下。透過尋找更有效率的參數組態來維持所需的效能等級，同時使用較少的資源，自動最佳化可找出降低成本的機會。

例如，自動最佳化可能會發現您目前的m值 （圖形連線能力） 高於準確度需求所需的值，可讓您減少索引時間和儲存體，而不會影響搜尋品質。

效能故障診斷

在向量搜尋操作中遇到查詢效能緩慢或高延遲時，自動最佳化可以分析資料集並識別更理想的組態。此服務提供特定建議來解決效能瓶頸，例如調整圖形連線或搜尋參數。

限制

帳單與成本

無論資料集大小和最佳化組態為何，自動最佳化都會使用每項工作的定價模型，讓您為每項成功的最佳化任務付費。您不需要為失敗或取消的任務付費。此外，自動最佳化會在與受管或無伺服器 OpenSearch 叢集不同的基礎設施上執行，因此不會影響現有叢集的資源使用率。

定價方式

自動最佳化成本與標準 OpenSearch Serverless 或 OpenSearch 受管網域運算和儲存成本分開計費。

如需定價資訊，請參閱 Amazon OpenSearch Service 定價。

將 JSONL 轉換為 Parquet

如果您的資料為 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 OpenSearch Service 功能搭配使用，協助您建置和最佳化向量搜尋應用程式：