加载中

使用 Python 摄取数据

本指南将向您介绍如何开始使用

  • 使用 Python 安全地连接到 Elastic Cloud 托管版或 Elastic Cloud Enterprise
  • 从您的应用程序将数据摄入到您的部署中
  • 搜索和修改您的数据

如果您是刚接触 Elastic Stack 的 Python 应用程序开发人员，本内容可以帮助您更轻松地入门。

可用的 Agent 技能

有一个技能可用于帮助 AI 智能体处理此主题。

详细了解适用于 Elastic 的 Agent 技能

获取此技能

所需时间：45 分钟

这些步骤适用于您现有的应用程序。如果您还没有，可以使用此处包含的示例创建一个。

python -m pip install elasticsearch
python -m pip install elasticsearch-async
		
# Elasticsearch 7.x
elasticsearch>=7.0.0,<8.0.0
  1. 获取免费试用.
  2. 登录 Elastic Cloud。
  3. 选择创建部署 (Create deployment)。
  4. 为您的部署命名。您可以保留所有其他设置的默认值。
  5. 选择 Create deployment（创建部署）并保存您的 Elastic 部署凭据。稍后您将需要这些凭据。
  6. 当部署就绪后，点击 Continue（继续），系统将显示 Setup guides（设置指南）页面。要继续前往部署主页，请点击 I’d like to do something else（我想做其他事情）。

不想订阅其他服务？您也可以通过 AWS、Azure 和 GCP 市场获取 Elastic Cloud Hosted。

  1. 登录 Elastic Cloud Enterprise 管理控制台。
  2. 选择创建部署 (Create deployment)。
  3. 为您的部署命名。您可以保留所有其他设置的默认值。
  4. 选择 Create deployment（创建部署）并保存您的 Elastic 部署凭据。稍后您将需要这些凭据。
  5. 当部署就绪后，点击 Continue（继续），系统将显示 Setup guides（设置指南）页面。要继续前往部署主页，请点击 I’d like to do something else（我想做其他事情）。

连接到 Elastic Cloud 托管版或 Elastic Cloud Enterprise 时，您需要使用 Cloud ID 来指定连接详情。通过进入 Kibana 主菜单并选择“管理 (Management) > 集成 (Integrations)”，然后选择“查看部署详情 (View deployment details)”来查找您的 Cloud ID。

为了连接、向其流式传输数据以及发出查询，您需要考虑身份验证。支持两种身份验证机制：API 密钥和基本身份验证。为了让您快速上手，我们将在此展示如何使用基本身份验证，但您也可以按照稍后的说明生成 API 密钥。API 密钥更安全，是生产环境的首选。

对于基本身份验证，请使用您之前复制的相同部署凭据（username 和 password 参数）以及 Cloud ID。通过进入 Kibana 主菜单并选择“管理 (Management) > 集成 (Integrations)”，然后选择“查看部署详情 (View deployment details)”来查找您的 Cloud ID。（如果您没有保存密码，可以重置密码。）

您首先需要创建并编辑一个包含您部署详情的 example.ini 文件

[ELASTIC]
cloud_id = DEPLOYMENT_NAME:CLOUD_ID_DETAILS
user = elastic
password = LONGPASSWORD

以下示例需要在交互模式下的 Python 解释器中输入。为了方便您复制示例，提示符已被移除，解释器的输出按原样显示。

❯ python3
Python 3.9.6 (default, Jun 29 2021, 05:25:02)
[Clang 12.0.5 (clang-1205.0.22.9)] on darwin
Type "help", "copyright", "credits" or "license" for more information.

from elasticsearch import Elasticsearch, helpers
import configparser

config = configparser.ConfigParser()
config.read('example.ini')
		
['example.ini']
>>>
		
es = Elasticsearch(
    cloud_id=config['ELASTIC']['cloud_id'],
    http_auth=(config['ELASTIC']['user'], config['ELASTIC']['password'])
)

现在，您可以通过返回有关部署的一些信息来确认您已连接到该部署

es.info()
		
{'name': 'instance-0000000000',
  'cluster_name': '747ab208fb70403dbe3155af102aef56',
  'cluster_uuid': 'IpgjkPkVQ5efJY-M9ilG7g',
  'version': {'number': '7.15.0', 'build_flavor': 'default', 'build_type': 'docker', 'build_hash': '79d65f6e357953a5b3cbcc5e2c7c21073d89aa29', 'build_date': '2021-09-16T03:05:29.143308416Z', 'build_snapshot': False, 'lucene_version': '8.9.0', 'minimum_wire_compatibility_version': '6.8.0', 'minimum_index_compatibility_version': '6.0.0-beta1'},
  'tagline': 'You Know, for Search'}

连接到部署后，您就可以准备索引和搜索数据了。让我们创建一个新索引，插入一些我们喜爱角色的引言，然后刷新索引以便进行搜索。刷新操作会使自上次刷新以来对索引执行的所有操作都可供搜索。

es.index(
 index='lord-of-the-rings',
 document={
  'character': 'Aragorn',
  'quote': 'It is not this day.'
 })
		
{'_index': 'lord-of-the-rings',
  '_type': '_doc',
  '_id': 'IanWEnwBg_mH2XweqDqg',
  '_version': 1,
  'result': 'created',
  '_shards': {'total': 2, 'successful': 1, 'failed': 0}, '_seq_no': 34, '_primary_term': 1}
		
es.index(
 index='lord-of-the-rings',
 document={
  'character': 'Gandalf',
  'quote': 'A wizard is never late, nor is he early.'
 })
		
{'_index': 'lord-of-the-rings',
  '_type': '_doc',
  '_id': 'IqnWEnwBg_mH2Xwezjpj',
  '_version': 1,
  'result': 'created',
  '_shards': {'total': 2, 'successful': 1, 'failed': 0}, '_seq_no': 35, '_primary_term': 1}
		
es.index(
 index='lord-of-the-rings',
 document={
  'character': 'Frodo Baggins',
  'quote': 'You are late'
 })
		
{'_index': 'lord-of-the-rings',
  '_type': '_doc',
  '_id': 'I6nWEnwBg_mH2Xwe_Tre',
  '_version': 1,
  'result': 'created',
  '_shards': {'total': 2, 'successful': 1, 'failed': 0}, '_seq_no': 36, '_primary_term': 1}
		
es.indices.refresh(index='lord-of-the-rings')
		
{'_shards': {'total': 2, 'successful': 1, 'failed': 0}}

使用 es.index API 时，如果 lord-of-the-rings 索引尚不存在，请求会自动创建该索引；如果未明确指定文档 ID，也会为每个被索引的文档创建文档 ID。

创建新索引并摄入一些数据后，您现在可以进行搜索了。让我们找出不同角色关于“迟到 (late)”说了些什么

result = es.search(
 index='lord-of-the-rings',
  query={
    'match': {'quote': 'late'}
  }
 )

result['hits']['hits']
		
[{'_index': 'lord-of-the-rings',
  '_type': '_doc',
  '_id': '2EkAzngB_pyHD3p65UMt',
  '_score': 0.5820575,
  '_source': {'character': 'Frodo Baggins', 'quote': 'You are late'}},
 {'_index': 'lord-of-the-rings',
  '_type': '_doc',
  '_id': '10kAzngB_pyHD3p65EPR',
  '_score': 0.37883914,
  '_source': {'character': 'Gandalf',
   'quote': 'A wizard is never late, nor is he early.'}}]

搜索请求返回在 quote 字段中包含 late 的文档内容，包括自动生成的文档 ID。

您可以使用文档 ID 对特定文档进行更新。让我们为我们的角色添加一个出生地

es.update(
 index='lord-of-the-rings',
 id='2EkAzngB_pyHD3p65UMt',
 doc={'birthplace': 'The Shire'}
 )
  1. 此更新示例使用 id 字段来标识要更新的文档。在更新和添加 birthplace 时，请复制与 Frodo Baggins 相关的文档中的 id。
es.get(index='lord-of-the-rings', id='2EkAzngB_pyHD3p65UMt')
{'_index': 'lord-of-the-rings',
 '_type': '_doc',
 '_id': '2EkAzngB_pyHD3p65UMt',
 '_version': 2,
 '_seq_no': 3,
 '_primary_term': 1,
 'found': True,
 '_source': {'character': 'Frodo Baggins',
  'quote': 'You are late',
  'birthplace': 'The Shire'}}

有关 Python 客户端的常用 API 调用，请查看示例。

一开始，对 Elasticsearch 的身份验证使用了 elastic 超级用户和密码，但使用 API 密钥会更安全，也是生产环境中的最佳实践。

在接下来的示例中，将创建一个具有集群 monitor 权限的 API 密钥，该权限提供用于确定集群状态的只读访问权限。一些额外的权限还允许对指定索引进行 create_index、write、read 和 manage 操作。添加了索引 manage 权限以启用索引刷新。

创建此密钥的最简单方法是在部署的 API 控制台中。选择部署名称并转到 ☰ > 管理 (Management) > 开发工具 (Dev Tools)

POST /_security/api_key
{
  "name": "python_example",
  "role_descriptors": {
    "python_read_write": {
      "cluster": ["monitor"],
      "index": [
        {
          "names": ["test-index"],
          "privileges": ["create_index", "write", "read", "manage"]
        }
      ]
    }
  }
}
		
{
  "id" : "API_KEY_ID",
  "name" : "python_example",
  "api_key" : "API_KEY_DETAILS"
}

编辑您之前创建的 example.ini 文件，并添加您刚刚创建的 id 和 api_key。在测试过 api_key 后，您还应该删除之前添加的 user 和 password 行，并考虑使用 Elastic Cloud 托管控制台或 Elastic Cloud Enterprise 管理控制台更改 elastic 密码。

[DEFAULT]
cloud_id = DEPLOYMENT_NAME:CLOUD_ID_DETAILS
apikey_id = API_KEY_ID
apikey_key = API_KEY_DETAILS

现在，您可以使用 API 密钥代替用户名和密码。客户端连接变为

es = Elasticsearch(
    cloud_id=config['DEFAULT']['cloud_id'],
    api_key=(config['DEFAULT']['apikey_id'], config['DEFAULT']['apikey_key']),
)

查看创建 API 密钥 API 以了解有关 API 密钥的更多信息，并查看安全权限以了解需要哪些权限。如果您不确定适合您的自定义应用程序的正确权限组合，可以在 Elasticsearch 上启用审计日志，以找出正在使用哪些权限。要了解有关 Elastic Cloud 托管版或 Elastic Cloud Enterprise 上日志记录工作原理的更多信息，请查看监控 Elastic Cloud 部署日志和指标。

有关刷新索引、搜索、更新和删除的更多信息，请查看 elasticsearch-py 示例。

安全

连接到 Elastic Cloud 托管版或 Elastic Cloud Enterprise 时，客户端默认会自动启用请求和响应压缩，因为这会显著提高吞吐量。此外，除非另有说明，客户端还会将 SSL 选项 secureProtocol 设置为 TLSv1_2_method。您仍然可以通过配置来覆盖此选项。

使用 Elastic Cloud 托管版或 Elastic Cloud Enterprise 时，请勿启用 sniffing（嗅探），因为节点位于负载均衡器之后。Elastic Cloud 托管版和 Elastic Cloud Enterprise 会为您处理好一切。如果您想了解更多信息，请查看 Elasticsearch sniffing（嗅探）最佳实践：什么、何时、为什么、如何。

架构
运行示例代码时，会自动创建索引映射。字段类型由 Elasticsearch 根据摄入第一条记录时看到的内容进行选择，并在数据中出现新字段时进行更新。预先指定字段和字段类型以优化性能会更有效。在为生产用例设计模式时，请参阅 Elastic Common Schema 文档和字段类型文档。
摄取 (Ingest)

对于更高级的场景，批量助手 (Bulk helpers) 提供了 bulk API 的示例，该 API 可以在单次调用中执行多个操作。如果您有大量文档需要索引，使用 bulk 来批量处理文档操作比单独提交请求要快得多。

© . This website operates independently and is not affiliated with or endorsed by Elasticsearch B.V. All brand names, logos, and trademarks are the property of their respective owners.