# 用 FSCrawler 和 Elasticsearch 构建知识库搜索引擎

> 使用 FSCrawler 将 PDF、Word 文档和扫描件索引到 Elasticsearch。涵盖 OCR、自定义映射和生产环境部署。

- 作者: zhuermu
- 发布: 2024-02-20
- 网页版: https://zhuermu.com/blog/fscrawler-elasticsearch-knowledge-base/
- 首发于: https://blog.csdn.net/qq258513813/article/details/136215593

---
每个组织都会积累大量文档——来自供应商的 PDF、各团队的 Word 报告、扫描的合同、会议上的幻灯片。这些内容承载着组织的知识资产，却被锁在任何搜索引擎都触及不到的文件里。Google 无法索引你的内部文件服务器，你的 wiki 搜索也读不懂一张扫描的发票。

**FSCrawler** 正是为解决这一问题而生。它监控一个目录（本地目录、远程目录，或通过 REST API 投递的文件），使用 Apache Tika 从任意文档格式中提取文本，可选地用 Tesseract 对扫描页面执行 OCR，并把所有内容索引到 Elasticsearch 以供全文检索。基础管线无需编写任何自定义代码——只需配置即可。

本文将从零开始搭建 FSCrawler，配置面向多语言文档的 OCR，构建自定义索引映射，用 Python 集成 REST API，并让整个系统在生产环境中运行。我们还会探讨 FSCrawler 在更宏观的知识库架构中的定位，以及它与 Apache Tika Server、Ingest Attachment 插件等替代方案的对比。

---

## 1. 架构概览

在动手安装之前，我们先来理解 FSCrawler 在知识库管线中的位置。

![知识库架构](/images/blog/fscrawler-elasticsearch-knowledge-base/architecture.svg)

整个架构分为四层：

1. **文件来源** —— 本地文件系统、挂载的网络驱动器、S3 存储桶、SSH/FTP 服务器，或通过 REST API 上传的文件。
2. **FSCrawler** —— 摄取引擎。它检测文件格式，用 Apache Tika 提取文本，对扫描文档执行 Tesseract OCR，并将所有内容批量索引到 Elasticsearch。
3. **Elasticsearch** —— 存储全文内容和元数据。通过 BM25 评分、过滤、高亮和聚合来处理搜索查询。
4. **搜索层** —— Kibana 的 Search Application 功能、自定义 REST API、Web 前端，或将结果喂给 LLM 的 RAG 管线。

这种关注点分离很重要。FSCrawler 不是搜索 UI，而是一条索引管线。你可以在不触碰摄取端的情况下更换搜索层，也可以在不改动搜索应用的情况下用其他索引器替换 FSCrawler。

---

## 2. 文档处理管线

下面是 FSCrawler 处理每个文件时发生的过程：

![文档处理管线](/images/blog/fscrawler-elasticsearch-knowledge-base/document-pipeline.svg)

1. **文件发现** —— FSCrawler 按固定间隔（可配置，默认 15 分钟）扫描配置的目录，检测新增文件、修改过的文件和已删除的文件。
2. **格式检测** —— Apache Tika 识别每个文件的 MIME 类型。
3. **文本提取** —— Tika 针对各种格式的解析器提取文本内容。它支持 PDF、DOC、DOCX、XLS、XLSX、PPT、PPTX、TXT、HTML、RTF 以及数十种其他格式。
4. **OCR（按需）** —— 如果文档是扫描版 PDF 或图片，且启用了 OCR，Tesseract 会从图像像素中提取文本。
5. **索引** —— 提取出的文本和元数据（文件名、路径、大小、内容类型、作者、创建日期、自定义标签）通过 Bulk API 发送到 Elasticsearch。

支持的文件格式包括：**PDF**（文本版和扫描版）、**DOC/DOCX**、**XLS/XLSX**、**PPT/PPTX**、**TXT**、**HTML**、**RTF**、**ODT**、**ODS**、**ODP**、**EPUB** 以及图片文件（通过 OCR）。

---

## 3. 使用 Docker 安装

FSCrawler 2.10 是当前的稳定版本。Docker 镜像是运行它最简单的方式——镜像中已经打包了 Java、Apache Tika 和 Tesseract OCR。

### 3.1 拉取镜像

```bash
docker pull dadoonet/fscrawler:2.10
```

### 3.2 创建工作目录

```bash
mkdir -p /data/fscrawler/config/job_name
mkdir -p /data/fscrawler/documents
```

目录结构如下：

```
/data/fscrawler/
├── config/
│   └── job_name/            # Job configuration directory
│       └── _settings.yaml   # Job settings (you create this)
└── documents/               # Files to be indexed
    ├── report.pdf
    ├── contract.docx
    └── presentation.pptx
```

### 3.3 运行 FSCrawler

```bash
docker run -it --rm \
  --name fscrawler \
  -v /data/fscrawler/config:/root/.fscrawler \
  -v /data/fscrawler/documents:/tmp/es:ro \
  dadoonet/fscrawler:2.10 fscrawler job_name
```

- `/root/.fscrawler` 是配置目录。FSCrawler 会从 job 子目录中读取 `_settings.yaml`。
- `/tmp/es:ro` 是文档目录（以只读方式挂载）。这里的所有文件都会被爬取并索引。
- `job_name` 是 job 标识符，同时也会成为默认的 Elasticsearch 索引名（文档用 `job_name`，文件夹条目用 `job_name_folder`）。

如果 `_settings.yaml` 不存在，FSCrawler 会在首次运行时创建一个默认配置。但只要不是玩具级 demo，你就应该自己编写配置文件。

---

## 4. 配置：`_settings.yaml`

所有重要的决策都体现在这里。下面是一份启用了 OCR、可直接用于生产环境的配置：

```yaml
---
name: "job_name"
fs:
  url: "/tmp/es"
  update_rate: "5m"
  excludes:
    - "*/~*"
    - "*/.DS_Store"
    - "*/Thumbs.db"
  json_support: false
  filename_as_id: false
  add_filesize: true
  remove_deleted: true
  store_source: false
  index_content: true
  index_folders: true
  lang_detect: false
  continue_on_error: true
  follow_symlinks: false
  ocr:
    language: "chi_sim+eng"
    enabled: true
    pdf_strategy: "ocr_and_text"
elasticsearch:
  nodes:
    - url: "https://your-elasticsearch-host:9200"
  api_key: "your-base64-encoded-api-key"
  bulk_size: 100
  flush_interval: "5s"
  byte_size: "10mb"
  ssl_verification: true
  push_templates: true
```

### 关键配置项说明

**`fs.update_rate`** —— FSCrawler 检查文件变化的频率。开发时设为 `1m`，生产环境设为 `5m` 到 `15m`。数值越低，I/O 负载越高。

**`fs.continue_on_error`** —— 生产环境请设为 `true`。单个损坏的文件不应中断整个爬取过程。

**`fs.ocr.language`** —— Tesseract 语言包。仅英文用 `eng`，简体中文加英文用 `chi_sim+eng`，或任意组合的 [Tesseract 语言代码](https://tesseract-ocr.github.io/tessdoc/Data-Files-in-different-versions.html)。

**`fs.ocr.pdf_strategy`** —— 控制 PDF 的处理方式：
- `"ocr_and_text"` —— 提取内嵌文本，并对图像页面执行 OCR。最适合混合型 PDF。
- `"ocr_only"` —— 只执行 OCR，忽略内嵌文本。适用于纯扫描文档。
- `"no_ocr"` —— 完全跳过 OCR。如果所有 PDF 都有内嵌文本，这是最快的选项。

**认证** —— FSCrawler 2.10 弃用了 `username`/`password`，改用 `api_key`。可以在 Kibana 的 **Stack Management > API Keys** 中生成 API key，或通过 Elasticsearch API 生成：

```bash
curl -X POST "https://your-es-host:9200/_security/api_key" \
  -H "Content-Type: application/json" \
  -u elastic:your-password \
  -d '{
    "name": "fscrawler-key",
    "role_descriptors": {
      "fscrawler_role": {
        "cluster": ["monitor"],
        "index": [
          {
            "names": ["job_name*"],
            "privileges": ["create_index", "write", "read", "manage"]
          }
        ]
      }
    }
  }'
```

响应中包含一个 `encoded` 字段——把它作为 `api_key` 的值即可。

---

## 5. 运行爬虫

### 5.1 首次运行

启动 FSCrawler 并观察日志：

```bash
docker run -it --rm \
  --name fscrawler \
  -v /data/fscrawler/config:/root/.fscrawler \
  -v /data/fscrawler/documents:/tmp/es:ro \
  dadoonet/fscrawler:2.10 fscrawler job_name
```

启动成功后，你会看到：

```
INFO  [f.p.e.c.f.FsCrawlerImpl] Starting FS crawler
INFO  [f.p.e.c.f.FsCrawlerImpl] FS crawler started in watch mode.
      It will run unless you stop it with CTRL+C.
INFO  [f.p.e.c.f.c.ElasticsearchClient] Elasticsearch Client connected
      to a node running version 8.17.0
INFO  [f.p.e.c.f.FsParserAbstract] FS crawler started for [job_name]
      for [/tmp/es] every [5m]
```

FSCrawler 会自动创建：
- 一个 `_default/` 目录，包含面向 6、7、8 版本的默认 Elasticsearch 索引模板。
- 一个 `_status.json` 文件，记录上次运行的时间戳：

```json
{
  "name": "job_name",
  "lastrun": "2024-02-21T07:55:58.851263972",
  "indexed": 28,
  "deleted": 0
}
```

### 5.2 理解文件同步行为

有两条重要的时序规则需要理解：

1. **初次同步** —— 在**首次**启动 FSCrawler**之前**，就把文件放进文档目录。这样能确保首次爬取时所有已有文件都被索引。
2. **增量同步** —— 首次运行之后，FSCrawler 只索引修改时间**晚于** `_status.json` 中 `lastrun` 时间戳的文件。如果你需要强制重新索引所有文件，删除 `_status.json` 后重启即可。

> **提示：** 如果你在首次运行之后添加历史文件却发现它们没被抓取，请检查它们的修改时间戳。你可能需要 `touch` 一下这些文件，或者删除 `_status.json`。

---

## 6. 在 Kibana 中验证

FSCrawler 运行之后，在 Kibana 中验证已索引的文档。

### 6.1 检查索引

在 Kibana 中进入 **Stack Management > Index Management**。你应该能看到两个索引：
- `job_name` —— 文档索引，包含提取出的内容和元数据。
- `job_name_folder` —— 文件夹索引（当 `index_folders: true` 时）。

### 6.2 通过 Dev Tools 查询文档

打开 Kibana 的 **Dev Tools** 并执行一次搜索：

```json
GET job_name/_search
{
  "query": {
    "match": {
      "content": "quarterly revenue"
    }
  },
  "_source": ["file.filename", "file.content_type", "file.filesize", "content"],
  "highlight": {
    "fields": {
      "content": {
        "fragment_size": 150,
        "number_of_fragments": 3
      }
    }
  }
}
```

### 6.3 在 Kibana 中创建 Search Application

Kibana 8.8+ 内置了 **Search Application** 功能，无需编写任何代码就能获得一个现成的搜索 UI：

1. 在侧边栏进入 **Enterprise Search > Search Applications**。
2. 点击 **Create**，选择你的 `job_name` 索引。
3. 给应用起个名字（例如 `knowledge-base`）。
4. 使用内置搜索 UI 测试查询——它开箱即用地展示文档内容、文件类型和相关度评分。

在投入开发自定义前端之前，这是向相关方演示系统的绝佳方式。

---

## 7. 自定义索引映射

FSCrawler 的默认映射足以应对基础搜索，但生产系统往往需要自定义分析器、额外字段或不同的字段类型。下面介绍如何自定义映射。

### 7.1 为什么要自定义？

- **自定义分析器** —— 使用面向特定语言的分析器（例如面向 CJK 文本的 `icu_analyzer`），而非默认的 standard 分析器。
- **keyword 字段** —— 将 `file.extension` 和 `file.content_type` 设为 keyword 字段，以支持精确匹配过滤和聚合。
- **额外字段** —— 添加业务元数据字段（部门、项目、密级）。
- **禁用 source 存储** —— 对大文档不存储 `_source` 以节省磁盘空间（仍可搜索，但无法取回原文）。

### 7.2 提供自定义映射

在 job 配置目录中创建文件 `_default/8/_settings_folder.json`（用于 ES 8.x）。下面是一个针对英文内容配置了自定义分析器的示例：

```json
{
  "settings": {
    "number_of_shards": 1,
    "number_of_replicas": 1,
    "analysis": {
      "analyzer": {
        "content_analyzer": {
          "type": "custom",
          "tokenizer": "standard",
          "filter": [
            "lowercase",
            "stop",
            "snowball",
            "asciifolding"
          ]
        }
      }
    }
  },
  "mappings": {
    "properties": {
      "content": {
        "type": "text",
        "analyzer": "content_analyzer",
        "fields": {
          "keyword": {
            "type": "keyword",
            "ignore_above": 256
          }
        }
      },
      "file": {
        "properties": {
          "content_type": { "type": "keyword" },
          "filename": {
            "type": "text",
            "fields": {
              "keyword": { "type": "keyword" }
            }
          },
          "extension": { "type": "keyword" },
          "filesize": { "type": "long" },
          "last_modified": { "type": "date" },
          "url": { "type": "keyword" }
        }
      },
      "path": {
        "properties": {
          "virtual": { "type": "keyword" },
          "real": { "type": "keyword" }
        }
      },
      "meta": {
        "properties": {
          "author": { "type": "text" },
          "title": { "type": "text" },
          "keywords": { "type": "keyword" }
        }
      },
      "external": {
        "type": "object",
        "dynamic": true
      }
    }
  }
}
```

在 `_settings.yaml` 中设置 `push_templates: true`，FSCrawler 就会在启动时把这份映射推送到 Elasticsearch。

### 7.3 面向 CJK（中日韩）内容的映射

如果你的文档包含 CJK 文本，请使用 ICU 分析插件：

```bash
# Install the ICU plugin on your Elasticsearch cluster
bin/elasticsearch-plugin install analysis-icu
```

然后在映射中使用 `icu_analyzer`：

```json
{
  "content": {
    "type": "text",
    "analyzer": "icu_analyzer"
  }
}
```

---

## 8. 用于文件上传的 REST API

FSCrawler 内置了一个 REST API，让你可以通过程序上传文件——当文件来自 Web 应用、CI 管线或 S3 事件触发时非常有用。

### 8.1 启用 REST API

启动 FSCrawler 时加上 `--rest`：

```bash
docker run -it --rm \
  --name fscrawler \
  -p 8080:8080 \
  -v /data/fscrawler/config:/root/.fscrawler \
  -v /data/fscrawler/documents:/tmp/es:ro \
  dadoonet/fscrawler:2.10 fscrawler job_name --rest
```

### 8.2 检查状态

```bash
curl http://localhost:8080/fscrawler
```

响应：

```json
{
  "ok": true,
  "version": "2.10",
  "elasticsearch": "8.17.0",
  "settings": {
    "name": "job_name",
    "fs": {
      "url": "/tmp/es",
      "update_rate": "5m"
    }
  }
}
```

### 8.3 上传文件

```bash
# Simple upload
curl -F "file=@report.pdf" "http://localhost:8080/fscrawler/_document"
```

响应：

```json
{
  "ok": true,
  "filename": "report.pdf",
  "url": "https://your-es-host:9200/job_name/_doc/abc123def456"
}
```

### 8.4 上传时附带自定义标签

创建一个包含业务元数据的 `tags.json` 文件：

```json
{
  "external": {
    "department": "engineering",
    "project": "knowledge-base",
    "classification": "internal",
    "uploaded_by": "api-service"
  }
}
```

带标签上传：

```bash
curl -F "file=@report.pdf" -F "tags=@tags.json" \
  "http://localhost:8080/fscrawler/_document"
```

`external` 对象会被合并进 Elasticsearch 文档，使其可被搜索和过滤。

### 8.5 Python 客户端

下面是一个可用于生产环境的 FSCrawler REST API Python 客户端：

```python
"""FSCrawler REST API client for programmatic document upload."""

import json
import logging
from pathlib import Path

import requests

logger = logging.getLogger(__name__)


class FSCrawlerClient:
    """Client for the FSCrawler REST API."""

    def __init__(self, base_url: str = "http://localhost:8080"):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()

    def health_check(self) -> dict:
        """Check FSCrawler status and connectivity."""
        resp = self.session.get(f"{self.base_url}/fscrawler")
        resp.raise_for_status()
        return resp.json()

    def upload_document(
        self,
        file_path: str | Path,
        tags: dict | None = None,
        index: str | None = None,
    ) -> dict:
        """
        Upload a document to FSCrawler for indexing.

        Args:
            file_path: Path to the file to upload.
            tags: Optional dict of custom metadata (stored under 'external').
            index: Optional index name override (defaults to job name).

        Returns:
            Response dict with 'ok', 'filename', and 'url' fields.
        """
        file_path = Path(file_path)
        if not file_path.exists():
            raise FileNotFoundError(f"File not found: {file_path}")

        url = f"{self.base_url}/fscrawler/_document"
        if index:
            url += f"?index={index}"

        files = {"file": (file_path.name, open(file_path, "rb"))}

        if tags:
            tags_content = json.dumps({"external": tags})
            files["tags"] = ("tags.json", tags_content, "application/json")

        resp = self.session.post(url, files=files)
        resp.raise_for_status()

        result = resp.json()
        if not result.get("ok"):
            raise RuntimeError(f"Upload failed: {result}")

        logger.info("Uploaded %s -> %s", file_path.name, result.get("url"))
        return result

    def upload_directory(
        self,
        directory: str | Path,
        extensions: list[str] | None = None,
        tags: dict | None = None,
        recursive: bool = True,
    ) -> list[dict]:
        """
        Upload all matching files in a directory.

        Args:
            directory: Path to the directory.
            extensions: File extensions to include (e.g., ['.pdf', '.docx']).
                        If None, uploads all files.
            tags: Optional metadata applied to all files.
            recursive: Whether to search subdirectories.

        Returns:
            List of upload results.
        """
        directory = Path(directory)
        pattern = "**/*" if recursive else "*"
        results = []

        for file_path in sorted(directory.glob(pattern)):
            if not file_path.is_file():
                continue
            if extensions and file_path.suffix.lower() not in extensions:
                continue

            try:
                result = self.upload_document(file_path, tags=tags)
                results.append(result)
            except Exception as e:
                logger.error("Failed to upload %s: %s", file_path, e)
                results.append({"ok": False, "filename": file_path.name, "error": str(e)})

        return results


# ── Usage example ────────────────────────────────────────────
if __name__ == "__main__":
    client = FSCrawlerClient("http://localhost:8080")

    # Check connectivity
    status = client.health_check()
    print(f"FSCrawler {status['version']} connected to ES {status['elasticsearch']}")

    # Upload a single file with tags
    result = client.upload_document(
        "quarterly-report.pdf",
        tags={
            "department": "finance",
            "quarter": "Q4-2024",
            "classification": "confidential",
        },
    )
    print(f"Indexed: {result['filename']} -> {result['url']}")

    # Batch upload a directory
    results = client.upload_directory(
        "/data/incoming/reports/",
        extensions=[".pdf", ".docx", ".xlsx"],
        tags={"source": "automated-upload", "batch": "2024-02-20"},
    )
    print(f"Uploaded {sum(1 for r in results if r['ok'])} / {len(results)} files")
```

---

## 9. 性能调优

FSCrawler 的默认设置偏保守。对于大规模文档集（成千上万个文件），调优必不可少。

### 9.1 Elasticsearch 批量写入设置

`_settings.yaml` 中的这些设置控制 FSCrawler 如何向 Elasticsearch 发送数据：

| 设置项 | 默认值 | 推荐值 | 说明 |
|---|---|---|---|
| `bulk_size` | 100 | 100-500 | 每个批量请求的文档数 |
| `flush_interval` | `"5s"` | `"5s"`-`"30s"` | 两次刷写之间的最大间隔 |
| `byte_size` | `"10mb"` | `"10mb"`-`"50mb"` | 批量请求的最大字节数 |

```yaml
elasticsearch:
  bulk_size: 200
  flush_interval: "10s"
  byte_size: "25mb"
```

增大 `bulk_size` 会减少发往 Elasticsearch 的 HTTP 请求数，但会增加内存占用。对于大文件（数 MB 的 PDF），应保持较低的 `bulk_size`，以免超出 `byte_size`。

### 9.2 OCR 性能

OCR 是整条管线中最慢的环节——慢一个数量级。单张扫描页面的 OCR 可能耗时 2-5 秒，而文本提取只需毫秒级。

**提升 OCR 性能的策略：**

- **不需要就禁用 OCR。** 如果所有文档都有内嵌文本，将 `ocr.enabled` 设为 `false`。
- **使用 `ocr_and_text` 策略**而非 `ocr_only`。这样有内嵌文本的页面可以被快速提取，只有基于图像的页面才触发 OCR。
- **限制 OCR 语言。** 每增加一个语言包都会拉长处理时间。除非你真的全都需要，否则用 `eng` 而不是 `chi_sim+eng+jpn+kor`。
- **为 OCR 密集型负载的 Docker 容器分配更多内存：**

```bash
docker run -it --rm \
  --memory=4g \
  -e JAVA_OPTS="-Xmx2g" \
  -v /data/fscrawler/config:/root/.fscrawler \
  -v /data/fscrawler/documents:/tmp/es:ro \
  dadoonet/fscrawler:2.10 fscrawler job_name
```

### 9.3 爬取频率与资源占用的权衡

`update_rate` 设置控制 FSCrawler 扫描文件目录的频率。设得太低（例如 `10s`）会导致文件系统被持续扫描；设得太高（例如 `1h`）则会延迟新文档的可用时间。

**参考准则：**
- 开发环境：`1m`
- 活跃的文档摄取：`5m`
- 稳定且偶有更新的知识库：`15m`-`1h`
- 结合 REST API 实现实时上传时：`30m`-`1h`（REST API 会立即索引，目录扫描只是一道兜底保险）

---

## 10. 生产环境部署

### 10.1 以守护进程方式运行

在生产环境中，以后台分离的 Docker 容器方式运行 FSCrawler，并开启自动重启：

```bash
docker run -d \
  --name fscrawler \
  --restart unless-stopped \
  --memory=4g \
  -e JAVA_OPTS="-Xmx2g" \
  -p 8080:8080 \
  -v /data/fscrawler/config:/root/.fscrawler \
  -v /data/fscrawler/documents:/tmp/es:ro \
  dadoonet/fscrawler:2.10 fscrawler job_name --rest
```

或使用 Docker Compose：

```yaml
# docker-compose.yml
services:
  fscrawler:
    image: dadoonet/fscrawler:2.10
    container_name: fscrawler
    restart: unless-stopped
    mem_limit: 4g
    environment:
      - JAVA_OPTS=-Xmx2g
    ports:
      - "8080:8080"
    volumes:
      - ./config:/root/.fscrawler
      - ./documents:/tmp/es:ro
    command: fscrawler job_name --rest
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8080/fscrawler"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 60s
```

### 10.2 健康检查与监控

使用 REST API 的健康检查端点进行监控：

```bash
# Simple health check for load balancers or container orchestrators
curl -sf http://localhost:8080/fscrawler | jq '.ok'
```

要进行更全面的监控，可跟踪以下 Elasticsearch 指标：

```bash
# Document count in the index
curl -s "https://your-es-host:9200/job_name/_count" | jq '.count'

# Index size on disk
curl -s "https://your-es-host:9200/job_name/_stats/store" | jq '.indices.job_name.total.store.size_in_bytes'

# Check _status.json for last run time
cat /data/fscrawler/config/job_name/_status.json | jq '.lastrun'
```

### 10.3 处理大规模文件集

对于数万个文件的初次索引：

1. **在启动 FSCrawler 之前暂存好文件。** 先把所有文件放进文档目录，再启动爬虫。这样可以避免批量摄取过程中增量扫描带来的开销。
2. **在批量导入期间增大 Elasticsearch 的刷新间隔：**

```bash
# Before bulk load — reduce indexing overhead
curl -X PUT "https://your-es-host:9200/job_name/_settings" \
  -H "Content-Type: application/json" \
  -d '{"index": {"refresh_interval": "60s"}}'

# After bulk load — restore normal refresh
curl -X PUT "https://your-es-host:9200/job_name/_settings" \
  -H "Content-Type: application/json" \
  -d '{"index": {"refresh_interval": "1s"}}'
```

3. **为不同目录使用多个 FSCrawler job。** 每个 job 独立运行，可针对不同的文件类型或 OCR 设置进行配置。

### 10.4 使用 Amazon OpenSearch

如果你更倾向于托管服务，FSCrawler 也能与 **Amazon OpenSearch**（AWS 对 Elasticsearch 的分支）配合使用。配置几乎完全相同：

```yaml
elasticsearch:
  nodes:
    - url: "https://your-domain.us-east-1.es.amazonaws.com"
  api_key: "your-opensearch-api-key"
  ssl_verification: true
  push_templates: true
```

对于 OpenSearch Serverless 集合，你需要使用基于 IAM 的认证。通过环境变量或 IAM 角色为 FSCrawler 容器配置 AWS 凭证，并使用相应的 OpenSearch 端点。

---

## 11. 工具对比

FSCrawler 并不是把文档索引进 Elasticsearch 的唯一方式。下面是它与各替代方案的对比：

| 特性 | FSCrawler | Tika Server | Ingest Attachment | Unstructured.io |
|---|---|---|---|---|
| **部署方式** | 独立部署（Docker） | 独立部署（Docker） | ES 插件 | 独立部署（Docker） |
| **文件监控** | 内置目录监控 | 无（仅 API） | 无（仅 API） | 无（仅 API） |
| **REST 上传 API** | 有 | 有 | 通过 ES Ingest API | 有 |
| **OCR 支持** | Tesseract（内置） | Tesseract（内置） | 无 | Tesseract + PaddleOCR |
| **Elasticsearch 集成** | 原生（直接索引） | 无（返回文本） | 原生（ingest pipeline） | 通过连接器 |
| **格式覆盖** | 1000+（通过 Tika） | 1000+（通过 Tika） | 有限子集 | 25+ 种格式 |
| **自定义元数据/标签** | 有（external 对象） | 无 | 有（ingest pipeline） | 有 |
| **增量同步** | 有（基于时间戳） | 无 | 无 | 无 |
| **搭建复杂度** | 低（配置文件） | 低（API 调用） | 中（管线配置） | 中（Python SDK） |
| **最适合** | 文件系统索引 | 仅文本提取 | 小规模、集群内 | AI/ML 管线、RAG |

**何时选择 FSCrawler：**
- 你需要索引一个文件目录，并在文件变化时保持索引同步。
- 你想要一个几乎零代码的开箱即用方案——只需 Docker 和一个 YAML 配置。
- 你需要对扫描文档提供 OCR 支持。

**何时选择替代方案：**
- **Tika Server** —— 你只需要文本提取，不需要 Elasticsearch 索引。索引由你的应用自行处理。
- **Ingest Attachment 插件** —— 你已经在使用 Elasticsearch ingest pipeline，希望一切都留在集群内。注意：不支持 OCR。
- **Unstructured.io** —— 你在构建 RAG 管线，需要结构化的文档解析（表格、标题、章节），而非扁平的纯文本提取。

---

## 12. 与 RAG 管线集成

FSCrawler 与 RAG 系统互补得很好。FSCrawler 负责文档摄取的"硬骨头"——格式检测、文本提取、OCR，而 Elasticsearch 存储处理结果。你的 RAG 管线随后查询 Elasticsearch，为 LLM 检索相关上下文。

一种典型的集成模式：

```python
from elasticsearch import Elasticsearch

es = Elasticsearch(
    "https://your-es-host:9200",
    api_key="your-api-key",
)


def search_knowledge_base(query: str, top_k: int = 5) -> list[dict]:
    """Search the FSCrawler-indexed knowledge base."""
    results = es.search(
        index="job_name",
        body={
            "query": {
                "multi_match": {
                    "query": query,
                    "fields": ["content", "file.filename^2", "meta.title^3"],
                    "type": "best_fields",
                }
            },
            "size": top_k,
            "_source": ["content", "file.filename", "file.content_type", "meta.title"],
            "highlight": {
                "fields": {"content": {"fragment_size": 300, "number_of_fragments": 3}}
            },
        },
    )

    documents = []
    for hit in results["hits"]["hits"]:
        doc = {
            "filename": hit["_source"].get("file", {}).get("filename"),
            "content_type": hit["_source"].get("file", {}).get("content_type"),
            "title": hit["_source"].get("meta", {}).get("title"),
            "score": hit["_score"],
            "content": hit["_source"].get("content", ""),
            "highlights": hit.get("highlight", {}).get("content", []),
        }
        documents.append(doc)

    return documents


# Use in a RAG pipeline
context_docs = search_knowledge_base("employee onboarding policy")
context = "\n\n---\n\n".join(
    f"[{doc['filename']}]\n{doc['content'][:2000]}" for doc in context_docs
)
# Feed 'context' into your LLM prompt...
```

这种模式让你兼得两者之长：FSCrawler 处理解析 50 种不同文件格式的脏活累活，而你的 RAG 管线只需一次简单查询，就能从 Elasticsearch 拿到干净的文本。

---

## 结语

FSCrawler 属于那种把一件事做到极致的工具：它接收数十种格式的文件，提取其中的文本内容（包括对扫描文档进行 OCR），并把所有内容索引进 Elasticsearch。无需自定义代码，无需复杂的管线编排——只要一个 Docker 容器和一份 YAML 配置文件。

关键要点：

1. **从 Docker 和一份简单的 `_settings.yaml` 起步。** 先让文档流入 Elasticsearch，再谈任何优化。
2. **只在需要时才启用 OCR。** 它是最大的单一性能瓶颈。对混合型文档集使用 `ocr_and_text` 策略。
3. **用 API key 而非用户名/密码。** `username`/`password` 字段在 FSCrawler 2.10 中已被弃用。
4. **为生产环境自定义索引映射。** 默认映射能用，但自定义分析器和 keyword 字段会显著提升搜索质量。
5. **使用 REST API** 进行程序化上传。与目录监控结合，就能同时覆盖批量摄取和实时摄取。
6. **用健康检查做监控**，并跟踪 `_status.json` 文件，以便及早发现爬取失败。

对于构建内部知识库、文档搜索系统，或 RAG 管线检索层的团队来说，FSCrawler 是一个坚实的基础，让你无需编写自定义的文档解析代码。

---

## 参考资料

- [FSCrawler 文档](https://fscrawler.readthedocs.io/)
- [FSCrawler GitHub 仓库](https://github.com/dadoonet/fscrawler)
- [Elasticsearch 全文搜索指南](https://www.elastic.co/guide/en/elasticsearch/reference/current/full-text-queries.html)
- [Apache Tika 支持的格式](https://tika.apache.org/2.9.1/formats.html)
- [Tesseract OCR 语言数据](https://tesseract-ocr.github.io/tessdoc/Data-Files-in-different-versions.html)
- [Kibana Search Applications](https://www.elastic.co/guide/en/kibana/current/search-applications.html)
- [Amazon OpenSearch 文档](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/)

---

## 常见问题

### FSCrawler 是什么？它如何把文档索引进 Elasticsearch？

FSCrawler 是一条文档摄取管线：它监控一个目录（本地、远程或通过 REST API 投递文件），用 Apache Tika 从任意文档格式中提取文本，可选地用 Tesseract 对扫描页面执行 OCR，再把文本和元数据批量索引到 Elasticsearch 供全文检索。基础管线无需编写任何自定义代码——只要一个 Docker 容器和一份 YAML 配置文件。

### 如何在 FSCrawler 中为扫描版 PDF 启用 OCR？

在 _settings.yaml 中把 fs.ocr.enabled 设为 true，通过 fs.ocr.language 选择 Tesseract 语言包（如 chi_sim+eng 表示简体中文加英文），并设置 pdf_strategy：ocr_and_text 提取内嵌文本并对图像页面执行 OCR（最适合混合型 PDF）；ocr_only 只做 OCR；no_ocr 完全跳过。注意 OCR 是整条管线中最慢的环节——单张扫描页可能耗时 2-5 秒，而文本提取只需毫秒级。

### 为什么首次运行之后新加入的文件没有被 FSCrawler 索引？

首次运行之后，FSCrawler 只索引修改时间晚于 _status.json 中 lastrun 时间戳的文件。如果你添加的历史文件时间戳较旧，就会被跳过。解决方法是用 touch 更新这些文件的修改时间，或者删除 _status.json 后重启 FSCrawler，强制对所有文件重新索引。


---

## 参考资料

- [FSCrawler documentation](https://fscrawler.readthedocs.io/) — Read the Docs
- [Elasticsearch reference](https://www.elastic.co/guide/en/elasticsearch/reference/current/index.html) — Elastic