关于

Eland 是一个 Python Elasticsearch 客户端（Elasticsearch 是一个分布式搜索和分析引擎），通过类 Pandas 的 API 探索和分析 Elasticsearch 中的数据。

在可能的情况下，该包使用现有的 Python API 和数据结构，使您能够轻松在 numpy、pandas 或 scikit-learn 与其 Elasticsearch 实现之间切换。通常，数据存储在 Elasticsearch 中而非内存中，这使 Eland 能够访问存储在 Elasticsearch 中的大型数据集。

Eland 还提供工具将来自常见库（如 scikit-learn、XGBoost 和 LightGBM）的训练好的机器学习模型上传到 Elasticsearch。

快速开始

Eland 可通过 Pip 从 PyPI 安装：

$ python -m pip install eland

如果使用 Eland 向 Elasticsearch 上传 NLP 模型，请安装 PyTorch 扩展：

$ python -m pip install 'eland[pytorch]'

Eland 也可通过 Conda 从 Conda Forge 安装：

$ conda install -c conda-forge eland

兼容性

• 支持 Python 3.10、3.11、3.12 和 3.13 版本
• 支持 Pandas 1.5 和 2 版本
• 支持 Elasticsearch 9+ 集群 如果使用 NLP 和 PyTorch 功能，请确保 Eland 的次版本号与 Elasticsearch 集群的次版本号匹配。对于其他功能，主版本号匹配即可。使用 Eland 8.x 对应 Elasticsearch 8.x
• 导入 NLP 模型需要安装适当版本的 PyTorch。运行 python -m pip install 'eland[pytorch]' 可安装对应版本

先决条件

基于 Debian 的系统用户可能需要安装 Eland 依赖的前置软件包：

$ sudo apt-get install -y \
  build-essential pkg-config cmake \
  python3-dev libzip-dev libjpeg-dev

注意：CentOS、RedHat、Arch 等其他发行版可能需要使用不同的包管理器和包名。

Docker

如需在不安装 Eland 的情况下运行脚本，可使用 Docker 镜像： 交互式运行：

$ docker run -it --rm --network host docker.elastic.co/eland/eland

无需交互式 shell 运行已安装的脚本示例：

$ docker run -it --rm --network host \
    docker.elastic.co/eland/eland \
    eland_import_hub_model \
      --url http://host.docker.internal:9200/ \
      --hub-model-id elastic/distilbert-base-cased-finetuned-conll03-english \
      --task-type ner

连接到 Elasticsearch

Eland 使用 Elasticsearch 低级客户端 连接 Elasticsearch。该客户端支持多种连接选项和认证选项。

您可以传递 elasticsearch.Elasticsearch 实例或包含连接地址的字符串给 Eland API：

import eland as ed

# 连接到本地运行的 Elasticsearch 实例
df = ed.DataFrame("http://localhost:9200", es_index_pattern="flights")

# 连接到 Elastic Cloud 实例
from elasticsearch import Elasticsearch

es = Elasticsearch(
    cloud_id="cluster-name:...",
    basic_auth=("elastic", "<password>")
)
df = ed.DataFrame(es, es_index_pattern="flights")

Eland 中的数据帧

eland.DataFrame 通过类 Pandas 的 API 封装 Elasticsearch 索引，将所有数据处理和过滤操作推迟到 Elasticsearch 而非本地机器执行。这意味着您可以通过 Jupyter Notebook 处理 Elasticsearch 中的大型数据集而不会使本地机器过载。

➤ Eland 数据帧 API 文档

➤ Jupyter Notebook 高级示例

>>> import eland as ed

>>> # 通过本地 Elasticsearch 节点连接 'flights' 索引
>>> df = ed.DataFrame('http://localhost:9200', 'flights')

# eland.DataFrame 实例的 API 与 pandas.DataFrame 相同
# 但所有数据存储在 Elasticsearch 中。查看 .info() 的内存使用情况
>>> df.head()
   AvgTicketPrice  Cancelled  ... dayOfWeek           timestamp
0      841.265642      False  ...         0 2018-01-01 00:00:00
1      882.982662      False  ...         0 2018-01-01 18:27:00
2      190.636904      False  ...         0 2018-01-01 17:11:14
3      181.694216       True  ...         0 2018-01-01 10:33:28
4      730.041778      False  ...         0 2018-01-01 05:13:00

[5 rows x 27 columns]

>>> df.info()
<class 'eland.dataframe.DataFrame'>
Index: 13059 条记录，从 0 到 13058
数据列 (共 27 列):
 #   列名              非空数量  数据类型         
---  ------              --------------  -----         
 0   AvgTicketPrice      13059 非空  float64       
 1   Cancelled           13059 非空  bool          
 2   Carrier             13059 非空  object        
...      
 24  OriginWeather       13059 非空  object        
 25  dayOfWeek           13059 非空  int64         
 26  timestamp           13059 非空  datetime64[ns]
数据类型：bool(2), datetime64[ns](1), float64(5), int64(2), object(17)
内存使用量：80.0 字节
Elasticsearch 存储使用量：5.043 MB

# 使用比较进行行过滤
>>> df[(df.Carrier=="Kibana Airlines") & (df.AvgTicketPrice > 900.0) & (df.Cancelled == True)].head()
     AvgTicketPrice  Cancelled  ... dayOfWeek           timestamp
8        960.869736       True  ...         0 2018-01-01 12:09:35
26       975.812632       True  ...         0 2018-01-01 15:38:32
311      946.358410       True  ...         0 2018-01-01 11:51:12
651      975.383864       True  ...         2 2018-01-03 21:13:17
950      907.836523       True  ...         2 2018-01-03 05:14:51

[5 rows x 27 columns]

# 在索引上执行聚合操作
>>> df[['DistanceKilometers', 'AvgTicketPrice']].aggregate(['sum', 'min', 'std'])
     DistanceKilometers  AvgTicketPrice
sum        9.261629e+07    8.204365e+06
min        0.000000e+00    1.000205e+02
std        4.578263e+03    2.663867e+02

Eland中的机器学习

回归与分类

Eland（爱兰德）支持将scikit-learn、XGBoost（极限梯度提升）和LightGBM（轻量级梯度提升机）库中训练好的回归和分类模型转换为可序列化的格式，并作为推理模型在Elasticsearch中使用。

➤ Eland机器学习API文档

➤ 了解Elasticsearch中的机器学习

>>> from sklearn import datasets
>>> from xgboost import XGBClassifier
>>> from eland.ml import MLModel

# 在本地训练并运行XGBoost机器学习模型
>>> training_data = datasets.make_classification(n_features=5)
>>> xgb_model = XGBClassifier(booster="gbtree")
>>> xgb_model.fit(training_data[0], training_data[1])

>>> xgb_model.predict(training_data[0])
[0 1 1 0 1 0 0 0 1 0]

# 将模型导入Elasticsearch
>>> es_model = MLModel.import_model(
    es_client="http://localhost:9200",
    model_id="xgb-classifier",
    model=xgb_model,
    feature_names=["f0", "f1", "f2", "f3", "f4"],
)

# 在Elasticsearch中使用训练数据运行机器学习模型
>>> es_model.predict(training_data[0])
[0 1 1 0 1 0 0 0 1 0]

使用PyTorch的NLP（自然语言处理）

[!WARNING]
PyTorch模型可能在Elasticsearch服务器上执行代码，使集群面临潜在安全漏洞风险。 仅使用可信来源的模型，切勿使用未经验证或未知提供者的模型。

对于自然语言处理任务，Eland支持将PyTorch训练的BERT模型导入Elasticsearch。模型可以是普通PyTorch模型，也可以是来自Hugging Face模型库的transformers（转换器）支持模型。

$ eland_import_hub_model \
  --url http://localhost:9200/ \
  --hub-model-id elastic/distilbert-base-cased-finetuned-conll03-english \
  --task-type ner \
  --start

上述示例将自动启动模型部署。这对于初始实验是很好的快捷方式，但对于需要良好吞吐量的场景，建议省略Eland命令行中的--start参数，改用Kibana的ML UI启动模型。--start参数将以每个分配一个线程的配置部署模型，性能表现不佳。通过Kibana的ML UI或ElasticsearchAPI启动模型部署时，可以设置线程选项以充分利用硬件性能。

>>> import elasticsearch
>>> from pathlib import Path
>>> from eland.common import es_version
>>> from eland.ml.pytorch import PyTorchModel
>>> from eland.ml.pytorch.transformers import TransformerModel

>>> es = elasticsearch.Elasticsearch("http://elastic:mlqa_admin@localhost:9200")
>>> es_cluster_version = es_version(es)

# 直接从模型库加载Hugging Face转换器模型
>>> tm = TransformerModel(model_id="elastic/distilbert-base-cased-finetuned-conll03-english", task_type="ner", es_version=es_cluster_version)
Downloading: 100%|██████████| 257/257 [00:00<00:00, 108kB/s]
Downloading: 100%|██████████| 954/954 [00:00<00:00, 372kB/s]
Downloading: 100%|██████████| 208k/208k [00:00<00:00, 668kB/s] 
Downloading: 100%|██████████| 112/112 [00:00<00:00, 43.9kB/s]
Downloading: 100%|██████████| 249M/249M [00:23<00:00, 11.2MB/s]

# 将模型导出为Elasticsearch使用的TorchScrpt格式
>>> tmp_path = "models"
>>> Path(tmp_path).mkdir(parents=True, exist_ok=True)
>>> model_path, config, vocab_path = tm.save(tmp_path)

# 将模型导入Elasticsearch
>>> ptm = PyTorchModel(es, tm.elasticsearch_model_id())
>>> ptm.import_model(model_path=model_path, config_path=None, vocab_path=vocab_path, config=config)
100%|██████████| 63/63 [00:12<00:00,  5.02it/s]

Eland 快速上手指南

环境准备

• Python版本：3.10/3.11/3.12/3.13
• 系统依赖（Debian/Ubuntu用户）：

  sudo apt-get install -y build-essential pkg-config cmake python3-dev libzip-dev libjpeg-dev
  

• Elasticsearch：9.x 或更新版本（需提前启动并配置）
• 可选依赖：使用NLP功能时需安装PyTorch（建议使用国内镜像安装）

安装步骤

使用 pip 安装（推荐清华源加速）

# 基础安装
python -m pip install eland -i https://pypi.tuna.tsinghua.edu.cn/simple

# 安装NLP扩展（含PyTorch）
python -m pip install 'eland[pytorch]' -i https://pypi.tuna.tsinghua.edu.cn/simple

使用 conda 安装

conda install -c conda-forge eland

Docker 快速体验

# 交互式运行
docker run -it --rm --network host docker.elastic.co/eland/eland

# 直接执行命令示例
docker run -it --rm --network host \
  docker.elastic.co/eland/eland \
  eland_import_hub_model \
    --url http://host.docker.internal:9200/ \
    --hub-model-id elastic/distilbert-base-cased-finetuned-conll03-english \
    --task-type ner

基本使用

连接Elasticsearch并读取数据

import eland as ed

# 连接本地Elasticsearch
df = ed.DataFrame("http://localhost:9200", es_index_pattern="flights")

# 查看前5行数据
df.head()

输出示例

   AvgTicketPrice  Cancelled  ... dayOfWeek           timestamp
0      841.265642      False  ...         0 2018-01-01 00:00:00
1      882.982662      False  ...         0 2018-01-01 18:27:00
2      190.636904      False  ...         0 2018-01-01 17:11:14
3      181.694216       True  ...         0 2018-01-01 10:33:28
4      730.041778      False  ...         0 2018-01-01 05:13:00

数据分析操作

# 查看数据统计信息
df.info()

# 条件过滤数据
df[(df.Carrier=="Kibana Airlines") & (df.AvgTicketPrice > 900.0)].head()

# 执行聚合计算
df[['DistanceKilometers', 'AvgTicketPrice']].aggregate(['sum', 'min', 'std'])

提示：首次使用建议先导入示例数据（如Elasticsearch自带的flights索引），具体可参考Elasticsearch官方示例数据导入指南。