> For the complete documentation index, see [llms.txt](https://docs.reorc.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.reorc.com/cn-reorc-help-center/data-ingestion/lian-jie-yuan-lei-xing/mongodb.md).

# MongoDB

`MongoDBDumpTask` 用于从 MongoDB 导出数据，但由于 MongoDB 的 schemaless 和特殊的查询语法等限制，目前仅提供有限的功能。 底层实现原理是调用 `find` 函数进行查询，详情可以参考 [MongoDB 官方 API 文档](https://api.mongodb.com/python/current/api/pymongo/collection.html#pymongo.collection.Collection.find)

#### Database

指定数据库名字

#### Collection

指定集合名字，一次只能操作一个集合

#### Filter

查询条件，是个 JSON 格式的字符串，支持 Jinja2 模板变量来实现一些动态查询条件（主要是时间）。 PyMongo 接受的 `filter` 是一个 `SON` 对象，OneFlow 无法直接配置，而是通过 JSON 来实现转换。 对于一些特殊类型，如 `ObjectId`, `datetime.datetime` JSON 不支持序列化，所以用了 `bson.json_util` 来序列化。 特殊类型配置方法如下，注意，如果数据库里保存的是时间类型（IOSDate），传入字符串格式的时间（`"2019-01-01 12:00:00"`）不生效。

* 传入 `ObjectId`

```json
{"_id": {"$gte": {"$oid": "595e0cfb6f6a6b1e40b3d6c5"}}}
```

* 传入时间，如 `dttm = datetime.datetime(2019, 7, 4, 13, 45, 16, 180975)`

```json
{"snapshot_time": {"$gte": {"$date": 1562247916180}}}   // 结果是时间戳，精确到毫秒 = dttm.timestamp() * 1000
```

* 传入模板变量

```json
{
  "snapshot_time": {
    "$gte": {"$date": {{ yesterday_dttm.timestamp * 1000 }}},
    "$lt": {"$date": {{ dttm.timestamp * 1000 }}},
   }
}
```

#### Projection

控制返回的字段，直接传入 find 函数，默认获取所有字段。具体用法参考 [MongoDB 官方文档](https://docs.mongodb.com/manual/tutorial/project-fields-from-query-results/)

#### Transform

定义 Python 代码对结果进行处理，MongoDB 支持嵌套结构，一般需要自定义解析逻辑来转换成单层结构。传入的 `row` 是个 `OrderedDict` 对象。 参考本文档对 Transform 的说明，请注意在 Load 里面传入建表语句 `Create Table DDL`.

#### 自动加上增量条件

`Incremental By Time`, `Time Column`, `Timezone`, `Round Time Resolution` 用于告诉 OneFlow 自动加上增量同步的条件。 效果等价于在 `Filter` 里通过变量传入时间范围。