从私有VPC内的AWS OpenSearch导出数据至本地容器

针对你遇到的API返回格式与_bulk要求不符的问题，这里提供几个实用的解决方法：

方法一：Python脚本转换格式并批量导入 #

因为你已经能通过API获取数据，最直接的方式是写个简单脚本，把返回的文档转换成_bulk要求的格式，再提交到本地OpenSearch。

步骤： #

1. 用Scroll API拉取全部数据：避免一次性拉取1万条数据导致超时，Scroll API适合批量迭代获取全量数据。
2. 生成_bulk兼容格式：每条数据需要拆成两行——第一行是操作指令（指定index、文档ID和目标索引），第二行是文档内容，每行单独成JSON，不能有换行。
3. 批量提交到本地OpenSearch：按批次（比如每1000条）发送_bulk请求。

示例代码： #

import requests

# 云端OpenSearch配置（确保本地能访问，比如通过VPC端口转发/VPN）
cloud_opensearch_url = "https://your-cloud-opensearch-endpoint"
cloud_index = "your-target-index"
scroll_id = None

# 本地OpenSearch配置
local_opensearch_url = "http://localhost:9200"
local_index = "your-local-index"

# 第一步：初始化Scroll获取第一批数据
init_response = requests.post(
    f"{cloud_opensearch_url}/{cloud_index}/_search?scroll=1m",
    json={"size": 1000, "query": {"match_all": {}}},
    auth=("your-cloud-username", "your-cloud-password")  # 如果云端开启了认证
)
init_data = init_response.json()
scroll_id = init_data["_scroll_id"]
hits = init_data["hits"]["hits"]

# 处理并导入数据的函数
def import_bulk_docs(docs):
    bulk_data = []
    for doc in docs:
        # 第一行：操作指令，复用原文档ID和目标索引
        bulk_data.append({"index": {"_index": local_index, "_id": doc["_id"]}})
        # 第二行：文档内容，取_source字段
        bulk_data.append(doc["_source"])
    # 转换为每行一个JSON的字符串格式
    bulk_str = "\n".join([str(item).replace("'", '"') for item in bulk_data]) + "\n"
    # 发送_bulk请求到本地
    response = requests.post(
        f"{local_opensearch_url}/_bulk",
        data=bulk_str,
        headers={"Content-Type": "application/x-ndjson"}
    )
    print(f"导入批次结果：{response.json()}")

# 处理第一批数据
import_bulk_docs(hits)

# 第二步：循环Scroll获取剩余数据
while len(hits) > 0:
    scroll_response = requests.post(
        f"{cloud_opensearch_url}/_search/scroll",
        json={"scroll": "1m", "scroll_id": scroll_id},
        auth=("your-cloud-username", "your-cloud-password")
    )
    scroll_data = scroll_response.json()
    hits = scroll_data["hits"]["hits"]
    if hits:
        import_bulk_docs(hits)
    scroll_id = scroll_data["_scroll_id"]

# 清理Scroll上下文
requests.delete(
    f"{cloud_opensearch_url}/_search/scroll",
    json={"scroll_id": scroll_id},
    auth=("your-cloud-username", "your-cloud-password")
)

方法二：用Logstash自动迁移 #

Logstash可以直接对接两个OpenSearch实例，自动处理数据格式转换和批量传输，无需手动写脚本。

步骤： #

1. 安装Logstash：本地安装对应版本的Logstash（尽量和云端OpenSearch版本匹配）。
2. 创建配置文件（比如opensearch-migrate.conf）：

input {
  opensearch {
    hosts => ["https://your-cloud-opensearch-endpoint"]
    index => "your-target-index"
    user => "your-cloud-username"
    password => "your-cloud-password"
    scroll => "1m"
    size => 1000
  }
}

output {
  opensearch {
    hosts => ["http://localhost:9200"]
    index => "your-local-index"
    document_id => "%{[@metadata][_id]}"  # 复用原文档ID
  }
}

3. 启动Logstash：运行命令bin/logstash -f opensearch-migrate.conf（Windows用bin\logstash.bat），等待数据迁移完成。

方法三：快照恢复（适合网络打通的场景） #

如果本地能通过VPN或专线访问云端VPC的OpenSearch，可以用快照功能迁移：

• 在云端OpenSearch创建快照仓库（比如关联S3存储桶），拍摄目标索引的快照。
• 在本地OpenSearch配置相同的快照仓库（需要配置S3访问权限），然后从快照恢复数据到本地索引。

注意事项： #

• 确保本地OpenSearch的版本与云端一致，避免兼容性问题。
• 云端是私有VPC，本地必须通过VPN、堡垒机端口转发等方式打通网络，才能访问云端的OpenSearch API。
• 批量导入时控制批次大小（比如1000条/批），避免触发请求超时或限流。

内容的提问来源于stack exchange，提问作者Anna