MongoDB

MongoDBDumpTask 用于从 MongoDB 导出数据，但由于 MongoDB 的 schemaless 和特殊的查询语法等限制，目前仅提供有限的功能。 底层实现原理是调用 find 函数进行查询，详情可以参考 MongoDB 官方 API 文档

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

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

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

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

• 传入模板变量

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

Projection

控制返回的字段，直接传入 find 函数，默认获取所有字段。具体用法参考 MongoDB 官方文档

Transform

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

自动加上增量条件

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