开发者社区 > 大数据与机器学习 > 大数据开发治理DataWorks > 正文

DataWorks中mongo 增量同步?

DataWorks中mongo 增量同步?

展开
收起
真的很搞笑 2023-09-27 07:55:58 129 0
1 条回答
写回答
取消 提交回答
  • 设置任务依赖实现参数传递:设置节点依赖关系,调度配置都设置10分钟调度
    image.png

    1、使用两个赋值节点定义时间戳格式的时间开始时间:
    参数:day=$[yyyy-mm-dd] start_time=$[hh24:mi:ss- 1/24/60*10]
    赋值语言选ODPS SQL:select UNIX_TIMESTAMP("unknown unknown");
    结束时间:
    参数:day=$[yyyy-mm-dd] end_time=$[hh24:mi:ss]
    赋值语言选ODPS SQL:select UNIX_TIMESTAMP("unknown unknown");
    image.png

    2、配置MongoDB同步节点添加本节点输入参数 start_time和end_time,取值自上游的两个赋值节点

    image.png

    MongoDB原始数据:脚本模式配置示例代码,源端create_time是double类型,存的时间戳。
    image.png

    "query": "{'create_time':{'$gte':unknown,'$lt':unknown}}",
    脚本配置示例{ "type": "job", "steps": [ { "stepType": "mongodb", "parameter": { "datasource": "ds1", "query": "{'create_time':{'$gte':unknown,'$lt':unknown}}", "column": [ { "name": "doc_id", "type": "STRING" }, { "name": "create_time", "type": "DOUBLE" }, { "name": "date_time", "type": "DATE" } ], "collectionName": "test1" }, "name": "Reader", "category": "reader" }, { "stepType": "odps", "parameter": { "partition": "", "truncate": false, "compress": false, "datasource": "odps_first", "column": [ "doc_id", "create_time", "date_time" ], "emptyAsNull": false, "table": "tablename" }, "name": "Writer", "category": "writer" } ], "version": "2.0", "order": { "hops": [ { "from": "Reader", "to": "Writer" } ] }, "setting": { "errorLimit": { "record": "" }, "speed": { "throttle": false, "concurrent": 2 } }},此回答整理自钉群“DataWorks交流群(答疑@机器人)”

    2023-09-27 14:51:05
    赞同 展开评论 打赏

DataWorks基于MaxCompute/Hologres/EMR/CDP等大数据引擎,为数据仓库/数据湖/湖仓一体等解决方案提供统一的全链路大数据开发治理平台。

相关产品

  • 大数据开发治理平台 DataWorks
  • 热门讨论

    热门文章

    相关电子书

    更多
    DataWorks调度任务迁移最佳实践-2020飞天大数据平台实战应用第一季 立即下载
    DataWorks商业化资源组省钱秘籍-2020飞天大数据平台实战应用第一季 立即下载
    基于DataWorks数据服务构建疫情大屏-2020飞天大数据平台实战应用第一季 立即下载

    相关实验场景

    更多