新手求助：如何从Azure Cognitive Search迁移数据至Elastic Search？

方案一：用 Logstash + HTTP Poller 插件实现迁移 #

虽然没有专门的Azure Cognitive Search Logstash插件，但可以用Logstash的http_poller输入插件调用Azure Cognitive Search的REST API拉取数据，再输出到Elasticsearch。

步骤说明 #

1. 配置HTTP Poller输入：通过Azure Cognitive Search的REST查询API批量拉取数据，需要用到你的搜索服务名称、索引名、管理员密钥。
2. 拆分返回的文档数组：Azure Search的API返回结果会把所有文档放在value数组里，用Logstash的split插件把每个文档拆成单独的事件。
3. 输出到Elasticsearch：配置Elasticsearch输出插件，指定目标集群地址和索引名。

示例Logstash配置文件（logstash.conf） #

input {
  http_poller {
    urls => {
      azure_search => "https://<你的搜索服务名>.search.windows.net/indexes/<你的索引名>/docs?api-version=2023-11-01&search=*&$top=1000"
    }
    request_timeout => 60
    interval => 300 # 按需调整拉取间隔，首次迁移可设为一次性执行
    method => "GET"
    headers => {
      "api-key" => "<你的Azure Search管理员密钥>"
      "Content-Type" => "application/json"
    }
    codec => "json"
  }
}

filter {
  # 拆分value数组中的每个文档
  split {
    field => "value"
    target => "document"
  }
  # 将文档内容移到根节点，方便Elasticsearch索引
  mutate {
    rename => { "[document][id]" => "id" }
    rename => { "[document][title]" => "title" }
    # 根据你的索引字段，继续添加需要映射的字段
    remove_field => ["value", "@timestamp", "@version"] # 移除不需要的字段
  }
}

output {
  elasticsearch {
    hosts => ["http://<Elasticsearch集群地址>:9200"]
    index => "<目标Elasticsearch索引名>"
    document_id => "%{id}" # 用Azure Search的文档ID作为Elasticsearch的文档ID
  }
  stdout { codec => rubydebug } # 可选，控制台输出调试
}

方案二：自定义Python脚本迁移（适合新手，更灵活） #

如果觉得Logstash配置麻烦，用Python脚本直接对接两个服务，代码逻辑更直观，适合一次性迁移或小规模数据。

步骤说明 #

1. 安装依赖库：

pip install azure-search-documents elasticsearch

2. 编写脚本，分页拉取Azure Search的数据，批量写入Elasticsearch。

示例Python脚本 #

from azure.search.documents import SearchClient
from azure.core.credentials import AzureKeyCredential
from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk

# 配置参数
AZURE_SEARCH_SERVICE = "<你的搜索服务名>"
AZURE_SEARCH_INDEX = "<你的索引名>"
AZURE_SEARCH_KEY = "<你的管理员密钥>"
ELASTICSEARCH_HOST = "<Elasticsearch集群地址>"
ELASTICSEARCH_INDEX = "<目标索引名>"

# 初始化Azure Search客户端
search_client = SearchClient(
    endpoint=f"https://{AZURE_SEARCH_SERVICE}.search.windows.net",
    index_name=AZURE_SEARCH_INDEX,
    credential=AzureKeyCredential(AZURE_SEARCH_KEY)
)

# 初始化Elasticsearch客户端
es_client = Elasticsearch(ELASTICSEARCH_HOST)

# 提前创建Elasticsearch索引（按需自定义字段映射）
# 对应Azure Search字段类型：Edm.String→text/keyword，Edm.Int32→integer，Edm.GeographyPoint→geo_point
es_client.indices.create(
    index=ELASTICSEARCH_INDEX,
    body={
        "mappings": {
            "properties": {
                "id": {"type": "keyword"},
                "title": {"type": "text"},
                "content": {"type": "text"},
                "location": {"type": "geo_point"}
                # 根据你的索引结构添加其他字段
            }
        }
    },
    ignore=400 # 索引已存在则忽略错误
)

# 分页拉取Azure Search数据
def fetch_azure_docs():
    continuation_token = None
    while True:
        results = search_client.search(search_text="*", top=1000, continuation_token=continuation_token)
        for doc in results:
            # 转换地理坐标格式（如果有）
            if "@search.geo_location" in doc:
                doc["location"] = {
                    "lat": doc["@search.geo_location"]["lat"],
                    "lon": doc["@search.geo_location"]["lon"]
                }
                del doc["@search.geo_location"]
            # 生成Elasticsearch批量写入格式
            yield {
                "_index": ELASTICSEARCH_INDEX,
                "_id": doc["id"],
                "_source": doc
            }
        continuation_token = results.continuation_token
        if not continuation_token:
            break

# 批量写入Elasticsearch
bulk(es_client, fetch_azure_docs())
print("迁移完成")

迁移注意事项 #

• 字段类型映射：Azure Search与Elasticsearch字段类型需手动对应，比如Edm.String对应Elasticsearch的text（全文搜索）或keyword（精确匹配），Edm.GeographyPoint对应geo_point。
• 分析器适配：如果Azure Search用了自定义分析器，需要在Elasticsearch中配置对应的分析器，保证搜索行为一致。
• 大数据量处理：严格按照分页拉取（Azure Search单次上限1000条），批量写入Elasticsearch，避免内存溢出。

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