亿级数据批量写入 Elasticsearch（简称 ES）是一个需要注意性能和可靠性的任务。以下是几种实现方式，每种方式都附有简要的描述和示例代码。
官方的 Elasticsearch 客户端库提供了丰富的功能和良好的性能，适用于大规模数据批量写入。你可以使用 bulk 方法来批量索引文档。
from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk
# 连接到Elasticsearch集群
es = Elasticsearch([{'host': 'localhost', 'port': 9200}])
# 构造要索引的数据
data_to_index = [
{"_index": "my_index", "_id": 1, "_source": {"field1": "value1"}},
{"_index": "my_index", "_id": 2, "_source": {"field1": "value2"}},
# 更多文档...
]
# 使用bulk方法批量写入数据
success, _ = bulk(es, data_to_index)
print(f"成功索引了{success}个文档")
Logstash 是一个数据处理工具，可以用来将各种数据源导入 Elasticsearch。它支持数据转换、过滤和批量写入。
配置文件示例 my_logstash.conf ：
input {
file {
path => "/path/to/data.json"
start_position => "beginning"
sincedb_path => "/dev/null"
}
}
output {
elasticsearch {
hosts => ["localhost:9200"]
index => "my_index"
}
}
然后运行 Logstash：
logstash -f my_logstash.conf
Elasticsearch 允许通过 HTTP 接口发送批量索引请求。
import requests
# Elasticsearch服务器信息
es_host = 'http://localhost:9200'
index_name = 'my_index'
data_to_index = [
{"index": {"_index": index_name, "_id": "1"}},
{"field1": "value1"},
{"index": {"_index": index_name, "_id": "2"}},
{"field1": "value2"},
# 更多文档...
]
# 构建批量索引请求
bulk_request = "\n".join([f'{{"{action}": {meta}}}\n{json}' for action, meta, json in data_to_index])
# 发送批量请求
response = requests.post(f"{es_host}/_bulk", data=bulk_request, headers={"Content-Type": "application/x-ndjson"})
print(response.content)
无论选择哪种方式，确保在批量写入过程中对性能和资源消耗进行监控，并采取适当的优化措施，如分批处理、调整 ES 集群配置等。
