devbox
小小林熬夜学编程
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

datax启动流程

作者：小小林熬夜学编程 | 2024-02-28 13:53:22

赞

踩

datax 进程

组件

datax采集流程
  • Reader：Reader为数据采集模块，负责采集数据源的数据，将数据发送给Framework。
  • Writer： Writer为数据写入模块，负责不断向Framework取数据，并将数据写入到目的端。
  • Transformer：在数据同步、传输过程中，存在用户对于数据传输进行特殊定制化的需求场景，包括裁剪列、转换列等工作
  • Job: Job是DataX用以描述从一个源头到一个目的端的同步作业，是DataX数据同步的最小业务单元。
  • Task: Task是把Job拆分得到的最小执行单元。
  • JobContainer: Job执行器，负责Job全局拆分、调度、前置语句和后置语句等工作的工作单元。
  • TaskGroupContainer: TaskGroup执行器，负责执行一组Task的工作单元。
  • TaskGroup: 描述的是一组Task集合。在同一个TaskGroupContainer执行下的Task集合称之为TaskGroup

参数

  1. datax.py脚本接收参数
  2. -j:jvm参数
  3. --jobid: 在local与distribute模式下运行的作业唯一id
  4. -m: 运行datax时的-Dmode参数，可选standalone, local, distribute 
  5. -p: 运行datax时的额外的附加的运行参数
  6. -r: 查看reader模板，与-w一起使用，${datax.home}/plugin/reader/${该读插件名称}/plugin_job_template.json
  7. -w: 查看writer模板，与-r一起使用，${datax.home}/plugin/reader/${该写插件名称}/plugin_job_template.json
  8.  
  9. C:/dev/Python27/python.exe datax.py -p"-Dlast=123 -Dend=456" --jobid=123456 C:/Users/Lenovo/Desktop/datax/jobConf/mysql2mysql.json
  10.  
  11. #分配了启动 限制堆大小为1g，不可扩展，发生了 内存溢出错误dump路径为C:\Users\Lenovo\PycharmProjects/log 
  12. java -server 
  13. -Xms1g -Xmx1g -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=C:\Users\Lenovo\PycharmProjects/log 
  14. -Dloglevel=info -Dfile.encoding=UTF-8 
  15. -Dlogback.statusListenerClass=ch.qos.logback.core.status.NopStatusListener 
  16. -Djava.security.egd=file:///dev/urandom 
  17. -Ddatax.home=C:\Users\Lenovo\PycharmProjects 
  18. -Dlogback.configurationFile=C:\Users\Lenovo\PycharmProjects/conf/logback.xml 
  19. -classpath C:\Users\Lenovo\PycharmProjects/lib/*  
  20. -Dlog.file.name=onf\mysql2mysql_json 
  21. -Dlast=123 -Dend=456 com.alibaba.datax.core.Engine 
  22. -mode standalone -jobid 123456 -job C:\Users\Lenovo\Desktop\datax\jobConf\mysql2mysql.json
  23. #默认以standalone模式启动
  24. datax.home=当前运行目录
  25. logback.configurationFile=${data.home}/conf/logback.xml 
  26. classpath=${datax.home}/lib/*

CoreConstant中会提取datax.home这个环境变量供全局使用，拼接成core.json，plugin.json的地址

  • 一些环境变量
  • mode：standalone, local, distribute 选择作业运行模式
  • jobid：在local与distribute模式下运行的作业唯一id
  • job：作业配置文件路径
  • classpath
  • Standalone: 单进程运行，没有外部依赖。
  • Local: 单进程运行，统计信息、错误信息汇报到集中存储。
  • Distrubuted: 分布式多进程运行，依赖DataX Service服务。

运行流程

引擎启动后jobContainer启动流程

作业配置加载

  • 通过作业配置文件路径(-job参数)来加载作业配置文件。

  • CoreConstant通过环境变量获取core配置文件路径(datax.home拼接)，加载core配置。

  • 通过job.content[0].reader/writer.name读取该作业的插件名，通过job.preHandler.pluginName/job.postHandler.pluginName读取该作业的前置或后置处理插件名。通过CoreConstant获取以上所有读取到的插件名的绝对路径。

  • 通过路径来加载插件配置文件内容。插件的配置文件按如下约束。

    1. {
    2.     "name": "mysqlwriter",
    3.     "class": "com.alibaba.datax.plugin.writer.mysqlwriter.MysqlWriter",
    4.     "description": "",
    5.     "developer": ""
    6. }

  • 在Configuration.from(String json)读取任意配置文件时都会将${xxx}或$xxx占位符替换成xxx对应的环境变量。即-Dlast=123使配置文件中 ${last}替换成123， 该逻辑存在StrUtil.replaceVariable(json)中

  • 将core，job，plugin配置合并，生成全局使用的配置Configuration。

    1. {
    2.     "entry":{……},
    3.     "common":{……},
    4.     "core":{
    5.         "container":{
    6.             "job":{
    7.                 "id": ${jobId},
    8.                 ………………其他配置
    9.           } 
    10.       }
    11.     }，
    12.     "job":{……},
    13.   "plugin":{
    14.         "reader":{
    15.             "${pluginName}":{
    16.                 "name": "",
    17.                 "class": "",
    18.                 "description": "",
    19.                 "developer": ""
    20.             }
    21.         },
    22.         "writer":{
    23.             "${pluginName}":{……}
    24.         }
    25.   }
    26. }

  • 最后做过滤输出和检查配置

引擎启动

  • 从common取出需要的转换格式yyyy-MM-dd或编码UTF-8，用于String与Date或Bytes的互相转换
  • 将配置Configuration传入LoadUtil Jar加载器，后面会使用LoadUtil进行插件Jar的动态加载。包括对每个插件的加载隔离机制和加载器缓冲的实现。
  • 根据core.container.model判断使用TaskGroupContainer还是使用JobContainer，默认使用JobContainer
  • PerfTrace初始化，默认不使用PerfTrace，获取job.JobInfo 默认无该配置项，
  • 容器启动

JobContainer

其中加载操作中的加载插件时：

  1. 为避免jar冲突，比如hbase可能有多个版本的读写依赖jar包，JobContainer和TaskGroupContainer
  2. 就需要脱离当前classLoader去加载这些jar包，执行完成后，又退回到原来classLoader上继续执行接下来的代码
  3.  
  4. 每个Jar的执行加载都有单一的类加载器进行隔离加载，JarLoader会缓冲到jarLoaderCenter中
  5.  
  6. LoadUtil.getJarLoader 就会根据插件类型和名字去jarLoaderCenter中获取加载器，获取不到之后才会重新构造一个加载器
类加载器隔离

  • preHandle前置处理器：根据job.preHandler.pluginName加载已存在的插件，并执行插件的preHandler方法

  • init初始化：

    1. 根据job.content[0].reader/writer.name插件名来加载reader和writer插件，并保存reader/wirter PluginName，

    2. 赋值Configuration，赋值Job插件本身和对端插件的配置job.content[0].reader/writer.parameter与对端的插件名子。并且执行他们的init方法。

  • prepare准备：执行reader/writer的prepare方法

  • split切分任务：

    1. 根据job.setting.speed.byte，core.transport.channel.speed.byte和job.setting.speed.record，core.transport.channel.speed.record的值计算出并发task数needChannelNumber，具体算法

      1. 作业byte限速除于单个channel的byte限速 得到 byte限速下的所需channel
      2. 作业record限速除于单个channel的record限速 得到record限速下的所需channel
      3. 对比两个channel数取最小的作为needChannelNumber
      4.  
      5. 若job.setting.speed.byte与job.setting.speed.record设置为空
      6. 则直接使用job.setting.speed.channel作为needChannelNumber
      7. 若都为空，则抛出异常

  1. 执行reader和writer的split方法，获取经过split每个Task的reader和writer的配置。

    1. 执行reader和writer最细粒度的切分，需要注意的是，writer的切分结果要参照reader的切分结果，
    2. 达到切分后数目相等，才能满足1：1的通道模型，所以这里可以将reader和writer的配置整合到一起。
    3.  
    4. 计算出的needChannelNumber/tableNUm * 分裂因子 = 最终需要的Task数量
切分方式 
在split方法中需要根据tables数量，splitPk进行分隔任务，每个任务下的connection都会根据切分结果与column，where来生成一个querySql。

  1. 获取作业的transformer配置，每个Task的reader和writer配置再加上该transformer的配置合并。将原本的job.content替换。即原本只有单个content，经过split后产生多个content，并为其设置递增的taskId

    1. {
    2.     "job": {
    3.         "content": [
    4.             {
    5.                 "taskId": 1,
    6.                 "reader": {
    7.                     "parameter": {
    8.                         "querySql": ""
    9.                     }
    10.                 },
    11.                 "transformer":[],
    12.                 "writer": {}
    13.             },
    14.              {
    15.                 "taskId": 2,
    16.                 "reader": {},
    17.                 "transformer":[],
    18.                 "writer": {}
    19.             },
    20.             {
    21.                 "taskId": 3,
    22.                 "reader": {},
    23.                 "transformer":[],
    24.                 "writer": {}
    25.             }
    26.         ],
    27.         "setting": {
    28.             "speed": {
    29.                 "channel": ""
    30.             }
    31.      }
    32.     }
    33. }

  • schedule调度：

    1. parseAndGetResourceMarkAndTaskIdMap：以reader.parameter.loadBalanceResourceMark资源名做分组。得出一个 资源名称 --> taskId(List) 的 map 映射关系。在split阶段，会对插件的loadBalanceResourceMark进行设置，通常是使用jdbc连接的host

    2. doAssign：根据parseAndGetResourceMarkAndTaskIdMap的结果，将需要运行Task按一个特定的规则分配到taskGroup中。每个TaskGroup都将获得一份Configuration克隆，设置每个taskConfiguration的content中的core.container.taskGroup.id。并且修正job.content，使他的配置文件回到单content状态

      1. a 库上有表：0, 1, 2
      2. b 库上有表：3, 4
      3. c 库上有表：5, 6, 7
      4.  
      5. 如果有 4个 taskGroup
      6. 打竖遍历添加到taskGroup 
      7. 资源: 0 3 5|1 4 6| 2 7
      8. taskGroup: 0 3 5 1|4 6 2 7
      9.  
      10. 则 doAssign 后的结果为：
      11. taskGroup-0: 0,  4,
      12. taskGroup-1: 3,  6,
      13. taskGroup-2: 5,  2,
      14. taskGroup-3: 1,  7

  1. adjustChannelNumPerTaskGroup：修正因为无法平均分配的少一个task的taskGroup的core.container.taskGroup.channel的更改

    1. 3个task分配到2个taskGroup中时，会造成一个taskGroup的channel为2,一个taskGroup的channel为1
    2. 所以要将少了一个task的taskGroup的channel进行修正优化。

  2. 为每个taskGroup修正core.container.job.mode为standalone

  3. StandAloneScheduler#registerCommunication：为每个taskGroup注册Communication(状态及统计信息交互)

  4. StandAloneScheduler#startAllTaskGroup：为每个taskGroup创建TaskGroupContainer并代理到TaskGroupContainerRunner启动TaskGroupContainer。其中动态加载transfomer，数据采集就在这个步骤之内。

  • post：执行reader和writer的post方法

  • postHandle：根据job.postHandler.pluginName加载已存在的插件，并执行插件的postHandler方法

  • invokeHooks：根据/hook目录调用外部hook

TaskGroupContainer

类图

reader与writer的数据传输

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小小林熬夜学编程/article/detail/159251
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号