提取、过滤和转换内容

连接器使用Elastic ingest attachment processor来提取文件内容。 处理器使用Apache Tika文本提取库来提取文件。 内容提取的逻辑定义在utils.py中。

虽然主要用于PDF和Microsoft Office格式，但您可以使用任何支持的格式。

企业搜索使用一个Elasticsearch 摄取管道来支持网络爬虫的二进制内容提取。 默认管道，ent-search-generic-ingestion，在企业搜索首次启动时自动创建。

您可以在 Kibana 中查看此管道。 自定义您的管道使用也是一个选项。 请参阅用于搜索索引的摄取管道。

对于高级用例，可以使用自托管提取服务来从大于10MB的文件中提取内容。

以下文件类型受支持：

标准内容提取通过附件处理器完成，使用 Elasticsearch Ingest Pipelines。 自管理连接器将管道提取的文件大小限制为每个文件 10MB（Elasticsearch 也有每个文件 100MB 的硬性限制）。

对于需要从超过这些限制的文件中提取内容的使用场景，可以使用自管理提取服务来自管理连接器。 与将文件作为附件发送到Elasticsearch不同，文件内容在摄取之前由提取服务在边缘提取。 提取的文本随后在摄取时作为文档的body字段包含在内。

要使用此功能，您需要执行以下操作：

本地内容提取适用于以下自托管连接器：

自托管的内容提取由一个独立的提取服务处理。

提取服务的版本与Elastic堆栈不一致。 对于8.11.x（包括9.0.0-beta1）之后的版本，您应该使用提取服务版本0.3.x。

您可以使用以下命令运行服务：

$ docker run \
  -p 8090:8090 \
  -it \
  --name extraction-service \
  docker.elastic.co/integrations/data-extraction-service:$EXTRACTION_SERVICE_VERSION

您可以通过添加所需的配置，使您的自托管连接器使用自托管提取服务。 自托管连接器通过配置文件中是否存在这些字段来确定是否启用了提取服务。

# data-extraction-service settings
extraction_service:
  host: http://localhost:8090

自托管连接器将在启动时对配置的host值执行预检。如果数据提取服务被找到并且正常运行，日志中将输出以下行。

Data extraction service found at <HOST>.

如果在启动时没有看到此日志，请参阅排查自托管内容提取服务问题。

以下字段可以包含在配置文件中。它们是可选的，如果未指定，将回退到默认值。

# data-extraction-service settings
extraction_service:
  host: http://localhost:8090
  timeout: 30
  use_file_pointers: false
  stream_chunk_size: 65536
  shared_volume_dir: '/app/files'

自托管的提取服务可以设置为使用文件指针而不是通过HTTP请求发送文件。 文件指针比发送文件更快且消耗更少的内存，但要求连接器框架和提取服务能够共享文件系统。 这可以通过docker化和非docker化的自管理连接器来设置。

如果您正在运行非Docker化的自托管连接器版本，您需要确定用于下载文件进行提取的本地目录。 这个目录可以是您文件系统中任何您觉得合适的位置。 请注意，自托管连接器将下载具有随机文件名的文件到此目录，因此任何已存在的文件都有可能被覆盖。 因此，我们建议为自托管提取使用一个专用的目录。

======= 示例

在使用从docker化的自管理连接器进行自托管提取时，除了在docker中运行自管理连接器之外，还需要一些额外的步骤。

======= 示例

提取服务生成两个不同的日志文件，这些文件在故障排除时可能会有所帮助。这些文件在docker容器内部保存在以下文件位置：

可以通过结合使用 docker exec 和 tail 命令从 Docker 外部查看日志。

$ docker exec extraction-service /bin/sh -c "tail /var/log/openresty.log"
$ docker exec extraction-service /bin/sh -c "tail /var/log/tika.log"

在使用自托管提取服务时，可能会出现以下警告日志。 本节中的每个日志后面都附有对其可能原因的描述以及建议的修复方法。

Extraction service is not configured, skipping its preflight check.

配置文件缺少extraction_service.host字段。 如果你想使用此服务，请检查配置是否格式正确，并确保所需字段存在。

Data extraction service found at <HOST>, but health-check returned <RESPONSE STATUS>.

/ping 端点返回了一个非 200 的响应。 这可能意味着提取服务不健康，可能需要重新启动，或者配置的 extraction_service.host 不正确。 您可以在 数据提取服务日志 中找到更多关于发生了什么的信息。

Expected to find a running instance of data extraction service at <HOST> but failed. <ERROR>.

健康检查返回了超时或客户端连接错误。

Extraction service has been initialised but no extraction service configuration was found. No text will be extracted for this sync.

您已为连接器启用了自托管提取服务，但配置文件中缺少 extraction_service.host 字段。 请检查配置是否格式正确，并确保所需字段存在。

Extraction service could not parse <FILENAME>. Status: <RESPONSE STATUS>; <ERROR NAME>: <ERROR MESSAGE>.

每次文件无法提取时，都会出现此警告。 通常， 会提供提取失败的原因说明。 如果消息不明确，请联系支持。 当文件提取失败时，它将在 body 字段中以空字符串进行索引。

使用连接器同步规则来帮助控制哪些文档在第三方数据源和Elasticsearch之间同步。 在Kibana UI中为每个连接器索引定义同步规则，位于索引的同步规则标签下。

同步规则适用于托管连接器和自管理连接器。

在 Elastic 版本 8.8.0 及更高版本 中，所有连接器都支持 基本 同步规则。

一些连接器支持高级同步规则。 了解更多信息，请参阅各个连接器的参考文档。

有两种同步规则：

在讨论同步规则之前，建立对数据过滤概念的基本理解是很重要的。 下图显示了数据过滤可以在几个不同的过程/位置发生。

在本文档中，我们将重点介绍远程和集成过滤。 同步规则可以用于修改这两者。

数据可能在源头被过滤。 我们称之为远程过滤，因为过滤过程是在 Elastic 之外进行的。

集成过滤充当原始数据源和Elasticsearch之间的桥梁。 在连接器中进行的过滤是集成过滤的一个例子。

最后，Elasticsearch 可以在数据持久化之前使用 ingest pipelines 进行过滤。 本指南中我们将不重点讨论 ingest pipeline 过滤。

在大多数情况下，您的数据湖将包含远比您希望向最终用户公开的数据更多的数据。 例如，您可能希望搜索产品目录，但不包括供应商联系信息，即使这两者出于业务目的而共存。

过滤数据的最佳时间是在数据管道的早期阶段。主要有两个原因：

在一个理想的世界中，所有的过滤操作都将在远程进行。

然而，在实践中，这并不总是可能的。 一些数据源不允许进行强大的远程过滤。 另一些则允许，但需要特殊的设置（在特定字段上建立索引、调整设置等），这可能需要您组织中其他利益相关者的关注。

考虑到这一点，同步规则被设计为修改远程过滤和集成过滤。 您的目标应该是尽可能多地进行远程过滤，但集成是一个完全可行的备用方案。 根据定义，远程过滤是在从第三方源获取数据之前应用的。 集成过滤是在从第三方源获取数据之后，但在将其摄取到Elasticsearch索引之前应用的。

每个基本同步规则可以是两种“策略”之一：include 和 exclude。 Include 规则用于包含“匹配”指定条件的文档。 Exclude 规则用于排除“匹配”指定条件的文档。

“匹配”是根据由“字段”、“规则”和“值”组合定义的条件来确定的。

应使用字段列来定义应考虑给定文档上的哪个字段。

以下规则可在规则列中使用：

最后，Value 列依赖于：

例如，对于一个regex规则，[A-Z]{2}这样的值可能是有意义的，但对于一个>规则来说，意义就不大了。 同样，当你处理一个ip_address字段时，你可能不会使用espresso这样的值，但对于一个beverage字段，你可能会使用。

排除所有在ID字段中值大于1000的文档。

排除所有在state字段中匹配指定正则表达式的文档。

以下图表说明了分页的概念。 一个庞大的数据集可能无法完全加载到连接器实例的内存中。 将数据分割成更小的块可以降低内存不足错误的风险。

此图展示了一次性提取整个数据集的过程：

相比之下，此图展示了一个分页数据集：

高级同步规则仅用于远程过滤。 你可以将高级同步规则视为一种与语言无关的方式来表示对数据源的查询。 因此，这些规则具有高度的源特定性。

以下连接器支持高级同步规则：

每个支持高级同步规则的连接器都提供了自己的DSL来指定规则。 请参阅每个连接器的文档以获取详细信息。

您还可以同时使用基本同步规则和高级同步规则来过滤数据集。

以下图表概述了高级同步规则、基本同步规则和管道过滤器应用于您的文档的顺序：

在下面的示例中，我们希望过滤一个包含公寓的数据集，使其仅包含具有特定属性的公寓。 我们将在整个示例中使用基本和高级同步规则。

一个示例公寓在 .json 格式中看起来像这样：

    {
        "id": 1234,
        "bedrooms": 3,
        "price": 1500,
        "address": {
            "street": "Street 123",
            "government_area": "Area",
            "country_information": {
                "country_code": "PT",
                "country": "Portugal"
    }
  }
}

目标数据集应满足以下条件：

前3个条件可以通过基本的同步规则来处理，但对于第4个条件，我们需要使用高级同步规则。

要创建一个新的基本同步规则，请导航到同步规则标签页并选择起草新的同步规则：

之后，您需要按下保存并验证草稿按钮来验证这些规则。 请注意，保存后规则将处于草稿状态。除非它们被应用，否则它们不会在下次同步时执行。

成功验证后，您可以应用您的规则，以便它们在下次同步时执行。

以下条件可以通过基本同步规则覆盖：

您只想包括位于葡萄牙或西班牙的公寓。 我们需要在这里使用高级同步规则，因为我们处理的是深度嵌套的对象。

假设公寓数据存储在MongoDB实例中。 对于MongoDB，我们在高级同步规则中支持聚合管道等功能。 一个仅选择位于葡萄牙或西班牙的房产的聚合管道如下所示：

    [
      {
        "$match": {
             "$or": [
                    {
                      "address.country_information.country": "Portugal"
                    },
                    {
                      "address.country_information.country": "Spain"
                    }
                  ]
                }
      }
    ]

要创建这些高级同步规则，请导航到同步规则创建对话框并选择高级规则选项卡。 您现在可以将聚合管道粘贴到aggregate.pipeline下的输入字段中：

一旦验证通过，应用这些规则。 以下截图显示了已应用的同步规则，这些规则将在下一次同步中执行：

成功同步后，您可以展开同步详情以查看应用了哪些规则：