安装完成，如果环境是Python3，则需要修改datax/bin下面的三个python文件。
linux安装DataX - 图1

GitHub

https://github.com/alibaba/DataX
https://github.com/alibaba/DataX/blob/master/introduction.md
https://github.com/alibaba/DataX/blob/master/userGuid.md

系统要求

  • Linux
  • JDK(1.8以上，推荐1.8)
  • Python(2或3都可以)
  • Apache Maven 3.x (Compile DataX)

    相关安装

    JDK安装

    https://www.oracle.com/java/technologies/downloads/archive/
    版本说明
    Linux x86 RPM Package //适用于32bit的centos、rethat（linux）操作系统
    Linux x64 RPM Package //适用于64bit的centos、rethat（linux）操作系统
    Linux x86 Compressed Archive //适用于32bit的Linux操作系统
    Linux x64 Compressed Archive //适用于64bit的Linux操作系统

    安装JDK

    1. mkdir /usr/local/jdk
    上传JDK后解压
    1. tar -zxvf jdk-8u341-linux-x64.tar.gz

    配置系统全局环境变量

    用vim编辑器来编辑profile文件，在文件末尾添加一下内容：
    如果没有vim，使用vi也可以，或者其他编辑器，都可以
    在文件中添加如下，根据自己实际情况填写
    1. vim /etc/profile
    1. export JAVA_HOME=/usr/local/jdk/jdk1.8.0_341/
    2. export CLASSPATH=.:$JAVA_HOME/lib:$JRE_HOME/lib:$CLASSPATH
    3. export PATH=$JAVA_HOME/bin:$JRE_HOME/bin:$PATH
    4. export JRE_HOME=$JAVA_HOME/jre
    :::warning JAVA_HOME为JDK 8安装目录 :::

    重新加载profile

    1. source /etc/profile

    查看版本

    1. java -version
    2. javac -version
    JDK 8安装完毕.

    安装 Python

    检查是否安装有Python
    1. python3 --version
    python 3.X版本在linux安装成功后,需要使用python2,不是python

    安装 DataX

    下载解压

    1. mkdir -p /usr/local/datax/ && cd /usr/local/datax/
    在线下载方式
    1. wget http://datax-opensource.oss-cn-hangzhou.aliyuncs.com/datax.tar.gz
    1. tar -zxvf datax.tar.gz -C /usr/local/
    手工下载方式: :::warning 直接下载DataX工具包：DataX下载地址,将下载的压缩包上传后解压即可。
    我解压后的存放目录为usr/local/datax :::

    删除隐藏文件

    需要删除隐藏文件 (重要)
    1. rm -rf /usr/local/datax/plugin/*/._*

    验证安装

    1. cd /usr/local/datax/bin
    1. python3 datax.py ../job/job.json
    输出 ```bash 2022-10-08 21:43:34.138 [job-0] INFO JobContainer - PerfTrace not enable! 2022-10-08 21:43:34.139 [job-0] INFO StandAloneJobContainerCommunicator - Total 100000 records, 2600000 bytes | Speed 253.91KB/s, 10000 records/s | Error 0 records, 0 bytes | All Task WaitWriterTime 0.071s | All Task WaitReaderTime 0.094s | Percentage 100.00% 2022-10-08 21:43:34.140 [job-0] INFO JobContainer - 任务启动时刻 : 2022-10-08 21:43:24 任务结束时刻 : 2022-10-08 21:43:34 任务总计耗时 : 10s 任务平均流量 : 253.91KB/s 记录写入速度 : 10000rec/s 读出记录总数 : 100000 读写失败总数 : 0
  1. <a name="xmSzz"></a>
  2. ## 修改python文件
  3. 安装完成，如果环境是Python3，则需要修改datax/bin下面的三个python文件。<br />[修改的文件](https://github.com/WeiYe-Jing/datax-web/tree/master/doc/datax-web/datax-python3)
  4. ```python
  5. #!/usr/bin/env python
  6. # -*- coding:utf-8 -*-
  8. import sys
  9. import os
  10. import signal
  11. import subprocess
  12. import time
  13. import re
  14. import socket
  15. import json
  16. from optparse import OptionParser
  17. from optparse import OptionGroup
  18. from string import Template
  19. import codecs
  20. import platform
  22. def isWindows():
  23.     return platform.system() == 'Windows'
  25. DATAX_HOME = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
  27. DATAX_VERSION = 'DATAX-OPENSOURCE-3.0'
  28. if isWindows():
  29.     codecs.register(lambda name: name == 'cp65001' and codecs.lookup('utf-8') or None)
  30.     CLASS_PATH = ("%s/lib/*") % (DATAX_HOME)
  31. else:
  32.     CLASS_PATH = ("%s/lib/*:.") % (DATAX_HOME)
  33. LOGBACK_FILE = ("%s/conf/logback.xml") % (DATAX_HOME)
  34. DEFAULT_JVM = "-Xms1g -Xmx1g -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=%s/log" % (DATAX_HOME)
  35. DEFAULT_PROPERTY_CONF = "-Dfile.encoding=UTF-8 -Dlogback.statusListenerClass=ch.qos.logback.core.status.NopStatusListener -Djava.security.egd=file:///dev/urandom -Ddatax.home=%s -Dlogback.configurationFile=%s" % (
  36.     DATAX_HOME, LOGBACK_FILE)
  37. ENGINE_COMMAND = "java -server ${jvm} %s -classpath %s  ${params} com.alibaba.datax.core.Engine -mode ${mode} -jobid ${jobid} -job ${job}" % (
  38.     DEFAULT_PROPERTY_CONF, CLASS_PATH)
  39. REMOTE_DEBUG_CONFIG = "-Xdebug -Xrunjdwp:transport=dt_socket,server=y,address=9999"
  41. RET_STATE = {
  42.     "KILL": 143,
  43.     "FAIL": -1,
  44.     "OK": 0,
  45.     "RUN": 1,
  46.     "RETRY": 2
  47. }
  50. def getLocalIp():
  51.     try:
  52.         return socket.gethostbyname(socket.getfqdn(socket.gethostname()))
  53.     except:
  54.         return "Unknown"
  57. def suicide(signum, e):
  58.     global child_process
  59.     print >> sys.stderr, "[Error] DataX receive unexpected signal %d, starts to suicide." % (signum)
  61.     if child_process:
  62.         child_process.send_signal(signal.SIGQUIT)
  63.         time.sleep(1)
  64.         child_process.kill()
  65.     print >> sys.stderr, "DataX Process was killed ! you did ?"
  66.     sys.exit(RET_STATE["KILL"])
  69. def register_signal():
  70.     if not isWindows():
  71.         global child_process
  72.         signal.signal(2, suicide)
  73.         signal.signal(3, suicide)
  74.         signal.signal(15, suicide)
  77. def getOptionParser():
  78.     usage = "usage: %prog [options] job-url-or-path"
  79.     parser = OptionParser(usage=usage)
  81.     prodEnvOptionGroup = OptionGroup(parser, "Product Env Options",
  82.                                      "Normal user use these options to set jvm parameters, job runtime mode etc. "
  83.                                      "Make sure these options can be used in Product Env.")
  84.     prodEnvOptionGroup.add_option("-j", "--jvm", metavar="<jvm parameters>", dest="jvmParameters", action="store",
  85.                                   default=DEFAULT_JVM, help="Set jvm parameters if necessary.")
  86.     prodEnvOptionGroup.add_option("--jobid", metavar="<job unique id>", dest="jobid", action="store", default="-1",
  87.                                   help="Set job unique id when running by Distribute/Local Mode.")
  88.     prodEnvOptionGroup.add_option("-m", "--mode", metavar="<job runtime mode>",
  89.                                   action="store", default="standalone",
  90.                                   help="Set job runtime mode such as: standalone, local, distribute. "
  91.                                        "Default mode is standalone.")
  92.     prodEnvOptionGroup.add_option("-p", "--params", metavar="<parameter used in job config>",
  93.                                   action="store", dest="params",
  94.                                   help='Set job parameter, eg: the source tableName you want to set it by command, '
  95.                                        'then you can use like this: -p"-DtableName=your-table-name", '
  96.                                        'if you have mutiple parameters: -p"-DtableName=your-table-name -DcolumnName=your-column-name".'
  97.                                        'Note: you should config in you job tableName with ${tableName}.')
  98.     prodEnvOptionGroup.add_option("-r", "--reader", metavar="<parameter used in view job config[reader] template>",
  99.                                   action="store", dest="reader",type="string",
  100.                                   help='View job config[reader] template, eg: mysqlreader,streamreader')
  101.     prodEnvOptionGroup.add_option("-w", "--writer", metavar="<parameter used in view job config[writer] template>",
  102.                                   action="store", dest="writer",type="string",
  103.                                   help='View job config[writer] template, eg: mysqlwriter,streamwriter')
  104.     parser.add_option_group(prodEnvOptionGroup)
  106.     devEnvOptionGroup = OptionGroup(parser, "Develop/Debug Options",
  107.                                     "Developer use these options to trace more details of DataX.")
  108.     devEnvOptionGroup.add_option("-d", "--debug", dest="remoteDebug", action="store_true",
  109.                                  help="Set to remote debug mode.")
  110.     devEnvOptionGroup.add_option("--loglevel", metavar="<log level>", dest="loglevel", action="store",
  111.                                  default="info", help="Set log level such as: debug, info, all etc.")
  112.     parser.add_option_group(devEnvOptionGroup)
  113.     return parser
  115. def generateJobConfigTemplate(reader, writer):
  116.     readerRef = "Please refer to the %s document:\n     https://github.com/alibaba/DataX/blob/master/%s/doc/%s.md \n" % (reader,reader,reader)
  117.     writerRef = "Please refer to the %s document:\n     https://github.com/alibaba/DataX/blob/master/%s/doc/%s.md \n " % (writer,writer,writer)
  118.     print (readerRef)
  119.     print (writerRef)
  120.     jobGuid = 'Please save the following configuration as a json file and  use\n     python {DATAX_HOME}/bin/datax.py {JSON_FILE_NAME}.json \nto run the job.\n'
  121.     print (jobGuid)
  122.     jobTemplate={
  123.       "job": {
  124.         "setting": {
  125.           "speed": {
  126.             "channel": ""
  127.           }
  128.         },
  129.         "content": [
  130.           {
  131.             "reader": {},
  132.             "writer": {}
  133.           }
  134.         ]
  135.       }
  136.     }
  137.     readerTemplatePath = "%s/plugin/reader/%s/plugin_job_template.json" % (DATAX_HOME,reader)
  138.     writerTemplatePath = "%s/plugin/writer/%s/plugin_job_template.json" % (DATAX_HOME,writer)
  139.     try:
  140.       readerPar = readPluginTemplate(readerTemplatePath);
  141.     except Exception as e:
  142.        print ("Read reader[%s] template error: can\'t find file %s" % (reader,readerTemplatePath))
  143.     try:
  144.       writerPar = readPluginTemplate(writerTemplatePath);
  145.     except Exception as e:
  146.       print ("Read writer[%s] template error: : can\'t find file %s" % (writer,writerTemplatePath))
  147.     jobTemplate['job']['content'][0]['reader'] = readerPar;
  148.     jobTemplate['job']['content'][0]['writer'] = writerPar;
  149.     print (json.dumps(jobTemplate, indent=4, sort_keys=True))
  151. def readPluginTemplate(plugin):
  152.     with open(plugin, 'r') as f:
  153.             return json.load(f)
  155. def isUrl(path):
  156.     if not path:
  157.         return False
  159.     assert (isinstance(path, str))
  160.     m = re.match(r"^http[s]?://\S+\w*", path.lower())
  161.     if m:
  162.         return True
  163.     else:
  164.         return False
  167. def buildStartCommand(options, args):
  168.     commandMap = {}
  169.     tempJVMCommand = DEFAULT_JVM
  170.     if options.jvmParameters:
  171.         tempJVMCommand = tempJVMCommand + " " + options.jvmParameters
  173.     if options.remoteDebug:
  174.         tempJVMCommand = tempJVMCommand + " " + REMOTE_DEBUG_CONFIG
  175.         print ('local ip: ', getLocalIp())
  177.     if options.loglevel:
  178.         tempJVMCommand = tempJVMCommand + " " + ("-Dloglevel=%s" % (options.loglevel))
  180.     if options.mode:
  181.         commandMap["mode"] = options.mode
  183.     # jobResource 鍙兘鏄?URL锛屼篃鍙兘鏄湰鍦版枃浠惰矾寰勶紙鐩稿,缁濆锛?    jobResource = args[0]
  184.     if not isUrl(jobResource):
  185.         jobResource = os.path.abspath(jobResource)
  186.         if jobResource.lower().startswith("file://"):
  187.             jobResource = jobResource[len("file://"):]
  189.     jobParams = ("-Dlog.file.name=%s") % (jobResource[-20:].replace('/', '_').replace('.', '_'))
  190.     if options.params:
  191.         jobParams = jobParams + " " + options.params
  193.     if options.jobid:
  194.         commandMap["jobid"] = options.jobid
  196.     commandMap["jvm"] = tempJVMCommand
  197.     commandMap["params"] = jobParams
  198.     commandMap["job"] = jobResource
  200.     return Template(ENGINE_COMMAND).substitute(**commandMap)
  203. def printCopyright():
  204.     print ('''
  205. DataX (%s), From Alibaba !
  206. Copyright (C) 2010-2017, Alibaba Group. All Rights Reserved.
  208. ''' % DATAX_VERSION)
  209.     sys.stdout.flush()
  212. if __name__ == "__main__":
  213.     printCopyright()
  214.     parser = getOptionParser()
  215.     options, args = parser.parse_args(sys.argv[1:])
  216.     if options.reader is not None and options.writer is not None:
  217.         generateJobConfigTemplate(options.reader,options.writer)
  218.         sys.exit(RET_STATE['OK'])
  219.     if len(args) != 1:
  220.         parser.print_help()
  221.         sys.exit(RET_STATE['FAIL'])
  223.     startCommand = buildStartCommand(options, args)
  224.     # print startCommand
  226.     child_process = subprocess.Popen(startCommand, shell=True)
  227.     register_signal()
  228.     (stdout, stderr) = child_process.communicate()
  230.     sys.exit(child_process.returncode)
  1. #! /usr/bin/env python
  2. # vim: set expandtab tabstop=4 shiftwidth=4 foldmethod=marker nu:
  4. import re
  5. import sys
  6. import time
  8. REG_SQL_WAKE = re.compile(r'Begin\s+to\s+read\s+record\s+by\s+Sql', re.IGNORECASE)
  9. REG_SQL_DONE = re.compile(r'Finished\s+read\s+record\s+by\s+Sql', re.IGNORECASE)
  10. REG_SQL_PATH = re.compile(r'from\s+(\w+)(\s+where|\s*$)', re.IGNORECASE)
  11. REG_SQL_JDBC = re.compile(r'jdbcUrl:\s*\[(.+?)\]', re.IGNORECASE)
  12. REG_SQL_UUID = re.compile(r'(\d+\-)+reader')
  13. REG_COMMIT_UUID = re.compile(r'(\d+\-)+writer')
  14. REG_COMMIT_WAKE = re.compile(r'begin\s+to\s+commit\s+blocks', re.IGNORECASE)
  15. REG_COMMIT_DONE = re.compile(r'commit\s+blocks\s+ok', re.IGNORECASE)
  17. # {{{ function parse_timestamp() #
  18. def parse_timestamp(line):
  19.     try:
  20.         ts = int(time.mktime(time.strptime(line[0:19], '%Y-%m-%d %H:%M:%S')))
  21.     except:
  22.         ts = 0
  24.     return ts
  26. # }}} #
  28. # {{{ function parse_query_host() #
  29. def parse_query_host(line):
  30.     ori = REG_SQL_JDBC.search(line)
  31.     if (not ori):
  32.         return ''
  34.     ori = ori.group(1).split('?')[0]
  35.     off = ori.find('@')
  36.     if (off > -1):
  37.         ori = ori[off+1:len(ori)]
  38.     else:
  39.         off = ori.find('//')
  40.         if (off > -1):
  41.             ori = ori[off+2:len(ori)]
  43.     return ori.lower()
  44. # }}} #
  46. # {{{ function parse_query_table() #
  47. def parse_query_table(line):
  48.     ori = REG_SQL_PATH.search(line)
  49.     return (ori and ori.group(1).lower()) or ''
  50. # }}} #
  52. # {{{ function parse_reader_task() #
  53. def parse_task(fname):
  54.     global LAST_SQL_UUID
  55.     global LAST_COMMIT_UUID
  56.     global DATAX_JOBDICT
  57.     global DATAX_JOBDICT_COMMIT
  58.     global UNIXTIME
  59.     LAST_SQL_UUID = ''
  60.     DATAX_JOBDICT = {}
  61.     LAST_COMMIT_UUID = ''
  62.     DATAX_JOBDICT_COMMIT = {}
  64.     UNIXTIME = int(time.time())
  65.     with open(fname, 'r') as f:
  66.         for line in f.readlines():
  67.             line = line.strip()
  69.             if (LAST_SQL_UUID and (LAST_SQL_UUID in DATAX_JOBDICT)):
  70.                 DATAX_JOBDICT[LAST_SQL_UUID]['host'] = parse_query_host(line)
  71.                 LAST_SQL_UUID = ''
  73.             if line.find('CommonRdbmsReader$Task') > 0:
  74.                 parse_read_task(line)
  75.             elif line.find('commit blocks') > 0:
  76.                 parse_write_task(line)
  77.             else:
  78.                 continue
  79. # }}} #
  81. # {{{ function parse_read_task() #
  82. def parse_read_task(line):
  83.     ser = REG_SQL_UUID.search(line)
  84.     if not ser:
  85.         return
  87.     LAST_SQL_UUID = ser.group()
  88.     if REG_SQL_WAKE.search(line):
  89.         DATAX_JOBDICT[LAST_SQL_UUID] = {
  90.             'stat' : 'R',
  91.             'wake' : parse_timestamp(line),
  92.             'done' : UNIXTIME,
  93.             'host' : parse_query_host(line),
  94.             'path' : parse_query_table(line)
  95.         }
  96.     elif ((LAST_SQL_UUID in DATAX_JOBDICT) and REG_SQL_DONE.search(line)):
  97.         DATAX_JOBDICT[LAST_SQL_UUID]['stat'] = 'D'
  98.         DATAX_JOBDICT[LAST_SQL_UUID]['done'] = parse_timestamp(line)
  99. # }}} #
  101. # {{{ function parse_write_task() #
  102. def parse_write_task(line):
  103.     ser = REG_COMMIT_UUID.search(line)
  104.     if not ser:
  105.         return
  107.     LAST_COMMIT_UUID = ser.group()
  108.     if REG_COMMIT_WAKE.search(line):
  109.         DATAX_JOBDICT_COMMIT[LAST_COMMIT_UUID] = {
  110.             'stat' : 'R',
  111.             'wake' : parse_timestamp(line),
  112.             'done' : UNIXTIME,
  113.         }
  114.     elif ((LAST_COMMIT_UUID in DATAX_JOBDICT_COMMIT) and REG_COMMIT_DONE.search(line)):
  115.         DATAX_JOBDICT_COMMIT[LAST_COMMIT_UUID]['stat'] = 'D'
  116.         DATAX_JOBDICT_COMMIT[LAST_COMMIT_UUID]['done'] = parse_timestamp(line)
  117. # }}} #
  119. # {{{ function result_analyse() #
  120. def result_analyse():
  121.     def compare(a, b):
  122.         return b['cost'] - a['cost']
  124.     tasklist = []
  125.     hostsmap = {}
  126.     statvars = {'sum' : 0, 'cnt' : 0, 'svr' : 0, 'max' : 0, 'min' : int(time.time())}
  127.     tasklist_commit = []
  128.     statvars_commit = {'sum' : 0, 'cnt' : 0}
  130.     for idx in DATAX_JOBDICT:
  131.         item = DATAX_JOBDICT[idx]
  132.         item['uuid'] = idx;
  133.         item['cost'] = item['done'] - item['wake']
  134.         tasklist.append(item);
  136.         if (not (item['host'] in hostsmap)):
  137.             hostsmap[item['host']] = 1
  138.             statvars['svr'] += 1
  140.         if (item['cost'] > -1 and item['cost'] < 864000):
  141.             statvars['sum'] += item['cost']
  142.             statvars['cnt'] += 1
  143.             statvars['max'] = max(statvars['max'], item['done'])
  144.             statvars['min'] = min(statvars['min'], item['wake'])
  146.     for idx in DATAX_JOBDICT_COMMIT:
  147.         itemc = DATAX_JOBDICT_COMMIT[idx]
  148.         itemc['uuid'] = idx
  149.         itemc['cost'] = itemc['done'] - itemc['wake']
  150.         tasklist_commit.append(itemc)
  152.         if (itemc['cost'] > -1 and itemc['cost'] < 864000):
  153.             statvars_commit['sum'] += itemc['cost']
  154.             statvars_commit['cnt'] += 1
  156.     ttl = (statvars['max'] - statvars['min']) or 1
  157.     idx = float(statvars['cnt']) / (statvars['sum'] or ttl)
  159.     tasklist.sort(compare)
  160.     for item in tasklist:
  161.         print '%s\t%s.%s\t%s\t%s\t% 4d\t% 2.1f%%\t% .2f' %(item['stat'], item['host'], item['path'],
  162.                                                            time.strftime('%H:%M:%S', time.localtime(item['wake'])),
  163.                                                            (('D' == item['stat']) and time.strftime('%H:%M:%S', time.localtime(item['done']))) or '--',
  164.                                                            item['cost'], 100 * item['cost'] / ttl, idx * item['cost'])
  166.     if (not len(tasklist) or not statvars['cnt']):
  167.         return
  169.     print '\n--- DataX Profiling Statistics ---'
  170.     print '%d task(s) on %d server(s), Total elapsed %d second(s), %.2f second(s) per task in average' %(statvars['cnt'],
  171.                                                                                                          statvars['svr'], statvars['sum'], float(statvars['sum']) / statvars['cnt'])
  172.     print 'Actually cost %d second(s) (%s - %s), task concurrency: %.2f, tilt index: %.2f' %(ttl,
  173.                                                                                              time.strftime('%H:%M:%S', time.localtime(statvars['min'])),
  174.                                                                                              time.strftime('%H:%M:%S', time.localtime(statvars['max'])),
  175.                                                                                              float(statvars['sum']) / ttl, idx * tasklist[0]['cost'])
  177.     idx_commit = float(statvars_commit['cnt']) / (statvars_commit['sum'] or ttl)
  178.     tasklist_commit.sort(compare)
  179.     print '%d task(s) done odps comit, Total elapsed %d second(s), %.2f second(s) per task in average, tilt index: %.2f' % (
  180.         statvars_commit['cnt'],
  181.         statvars_commit['sum'], float(statvars_commit['sum']) / statvars_commit['cnt'],
  182.         idx_commit * tasklist_commit[0]['cost'])
  184. # }}} #
  186. if (len(sys.argv) < 2):
  187.     print "Usage: %s filename" %(sys.argv[0])
  188.     quit(1)
  189. else:
  190.     parse_task(sys.argv[1])
  191.     result_analyse()
  1. #!/usr/bin/env python
  2. # -*- coding:utf-8 -*-
  5. """
  6.    Life's short, Python more.
  7. """
  9. import re
  10. import os
  11. import sys
  12. import json
  13. import uuid
  14. import signal
  15. import time
  16. import subprocess
  17. from optparse import OptionParser
  18. reload(sys)
  19. sys.setdefaultencoding('utf8')
  21. ##begin cli & help logic
  22. def getOptionParser():
  23.     usage = getUsage()
  24.     parser = OptionParser(usage = usage)
  25.     #rdbms reader and writer
  26.     parser.add_option('-r', '--reader', action='store', dest='reader', help='trace datasource read performance with specified !json! string')
  27.     parser.add_option('-w', '--writer', action='store', dest='writer', help='trace datasource write performance with specified !json! string')
  29.     parser.add_option('-c', '--channel',  action='store', dest='channel', default='1', help='the number of concurrent sync thread, the default is 1')
  30.     parser.add_option('-f', '--file',   action='store', help='existing datax configuration file, include reader and writer params')
  31.     parser.add_option('-t', '--type',   action='store', default='reader', help='trace which side\'s performance, cooperate with -f --file params, need to be reader or writer')
  32.     parser.add_option('-d', '--delete',   action='store', default='true', help='delete temporary files, the default value is true')
  33.     #parser.add_option('-h', '--help',   action='store', default='true', help='print usage information')
  34.     return parser
  36. def getUsage():
  37.     return '''
  38. The following params are available for -r --reader:
  39.     [these params is for rdbms reader, used to trace rdbms read performance, it's like datax's key]
  40.     *datasourceType: datasource type, may be mysql|drds|oracle|ads|sqlserver|postgresql|db2 etc...
  41.     *jdbcUrl:        datasource jdbc connection string, mysql as a example: jdbc:mysql://ip:port/database
  42.     *username:       username for datasource
  43.     *password:       password for datasource
  44.     *table:          table name for read data
  45.     column:          column to be read, the default value is ['*']
  46.     splitPk:         the splitPk column of rdbms table
  47.     where:           limit the scope of the performance data set
  48.     fetchSize:       how many rows to be fetched at each communicate
  50.     [these params is for stream reader, used to trace rdbms write performance]
  51.     reader-sliceRecordCount:  how man test data to mock(each channel), the default value is 10000
  52.     reader-column          :  stream reader while generate test data(type supports: string|long|date|double|bool|bytes; support constant value and random function)锛宒emo: [{"type":"string","value":"abc"},{"type":"string","random":"10,20"}]
  54. The following params are available for -w --writer:
  55.     [these params is for rdbms writer, used to trace rdbms write performance, it's like datax's key]
  56.     datasourceType:  datasource type, may be mysql|drds|oracle|ads|sqlserver|postgresql|db2|ads etc...
  57.     *jdbcUrl:        datasource jdbc connection string, mysql as a example: jdbc:mysql://ip:port/database
  58.     *username:       username for datasource
  59.     *password:       password for datasource
  60.     *table:          table name for write data
  61.     column:          column to be writed, the default value is ['*']
  62.     batchSize:       how many rows to be storeed at each communicate, the default value is 512
  63.     preSql:          prepare sql to be executed before write data, the default value is ''
  64.     postSql:         post sql to be executed end of write data, the default value is ''
  65.     url:             required for ads, pattern is ip:port
  66.     schme:           required for ads, ads database name
  68.     [these params is for stream writer, used to trace rdbms read performance]
  69.     writer-print:           true means print data read from source datasource, the default value is false
  71. The following params are available global control:
  72.     -c --channel:    the number of concurrent tasks, the default value is 1
  73.     -f --file:       existing completely dataX configuration file path
  74.     -t --type:       test read or write performance for a datasource, couble be reader or writer, in collaboration with -f --file
  75.     -h --help:       print help message
  77. some demo:
  78. perftrace.py --channel=10 --reader='{"jdbcUrl":"jdbc:mysql://127.0.0.1:3306/database", "username":"", "password":"", "table": "", "where":"", "splitPk":"", "writer-print":"false"}'
  79. perftrace.py --channel=10 --writer='{"jdbcUrl":"jdbc:mysql://127.0.0.1:3306/database", "username":"", "password":"", "table": "", "reader-sliceRecordCount": "10000", "reader-column": [{"type":"string","value":"abc"},{"type":"string","random":"10,20"}]}'
  80. perftrace.py --file=/tmp/datax.job.json --type=reader --reader='{"writer-print": "false"}'
  81. perftrace.py --file=/tmp/datax.job.json --type=writer --writer='{"reader-sliceRecordCount": "10000", "reader-column": [{"type":"string","value":"abc"},{"type":"string","random":"10,20"}]}'
  83. some example jdbc url pattern, may help:
  84. jdbc:oracle:thin:@ip:port:database
  85. jdbc:mysql://ip:port/database
  86. jdbc:sqlserver://ip:port;DatabaseName=database
  87. jdbc:postgresql://ip:port/database
  88. warn: ads url pattern is ip:port
  89. warn: test write performance will write data into your table, you can use a temporary table just for test.
  90. '''
  92. def printCopyright():
  93.     DATAX_VERSION = 'UNKNOWN_DATAX_VERSION'
  94.     print '''
  95. DataX Util Tools (%s), From Alibaba !
  96. Copyright (C) 2010-2016, Alibaba Group. All Rights Reserved.''' % DATAX_VERSION
  97.     sys.stdout.flush()
  100. def yesNoChoice():
  101.     yes = set(['yes','y', 'ye', ''])
  102.     no = set(['no','n'])
  103.     choice = raw_input().lower()
  104.     if choice in yes:
  105.         return True
  106.     elif choice in no:
  107.         return False
  108.     else:
  109.         sys.stdout.write("Please respond with 'yes' or 'no'")
  110. ##end cli & help logic
  113. ##begin process logic
  114. def suicide(signum, e):
  115.     global childProcess
  116.     print >> sys.stderr, "[Error] Receive unexpected signal %d, starts to suicide." % (signum)
  117.     if childProcess:
  118.         childProcess.send_signal(signal.SIGQUIT)
  119.         time.sleep(1)
  120.         childProcess.kill()
  121.     print >> sys.stderr, "DataX Process was killed ! you did ?"
  122.     sys.exit(-1)
  125. def registerSignal():
  126.     global childProcess
  127.     signal.signal(2, suicide)
  128.     signal.signal(3, suicide)
  129.     signal.signal(15, suicide)
  132. def fork(command, isShell=False):
  133.     global childProcess
  134.     childProcess = subprocess.Popen(command, shell = isShell)
  135.     registerSignal()
  136.     (stdout, stderr) = childProcess.communicate()
  137.     #闃诲鐩村埌瀛愯繘绋嬬粨鏉?    childProcess.wait()
  138.     return childProcess.returncode
  139. ##end process logic
  142. ##begin datax json generate logic
  143. #warn: if not '': -> true;   if not None: -> true
  144. def notNone(obj, context):
  145.     if not obj:
  146.         raise Exception("Configuration property [%s] could not be blank!" % (context))
  148. def attributeNotNone(obj, attributes):
  149.     for key in attributes:
  150.         notNone(obj.get(key), key)
  152. def isBlank(value):
  153.     if value is None or len(value.strip()) == 0:
  154.         return True
  155.     return False
  157. def parsePluginName(jdbcUrl, pluginType):
  158.     import re
  159.     #warn: drds
  160.     name = 'pluginName'
  161.     mysqlRegex = re.compile('jdbc:(mysql)://.*')
  162.     if (mysqlRegex.match(jdbcUrl)):
  163.         name = 'mysql'
  164.     postgresqlRegex = re.compile('jdbc:(postgresql)://.*')
  165.     if (postgresqlRegex.match(jdbcUrl)):
  166.         name = 'postgresql'
  167.     oracleRegex = re.compile('jdbc:(oracle):.*')
  168.     if (oracleRegex.match(jdbcUrl)):
  169.         name = 'oracle'
  170.     sqlserverRegex = re.compile('jdbc:(sqlserver)://.*')
  171.     if (sqlserverRegex.match(jdbcUrl)):
  172.         name = 'sqlserver'
  173.     db2Regex = re.compile('jdbc:(db2)://.*')
  174.     if (db2Regex.match(jdbcUrl)):
  175.         name = 'db2'
  176.     return "%s%s" % (name, pluginType)
  178. def renderDataXJson(paramsDict, readerOrWriter = 'reader', channel = 1):
  179.     dataxTemplate = {
  180.         "job": {
  181.             "setting": {
  182.                 "speed": {
  183.                     "channel": 1
  184.                 }
  185.             },
  186.             "content": [
  187.                 {
  188.                     "reader": {
  189.                         "name": "",
  190.                         "parameter": {
  191.                             "username": "",
  192.                             "password": "",
  193.                             "sliceRecordCount": "10000",
  194.                             "column": [
  195.                                 "*"
  196.                             ],
  197.                             "connection": [
  198.                                 {
  199.                                     "table": [],
  200.                                     "jdbcUrl": []
  201.                                 }
  202.                             ]
  203.                         }
  204.                     },
  205.                     "writer": {
  206.                         "name": "",
  207.                         "parameter": {
  208.                             "print": "false",
  209.                             "connection": [
  210.                                 {
  211.                                     "table": [],
  212.                                     "jdbcUrl": ''
  213.                                 }
  214.                             ]
  215.                         }
  216.                     }
  217.                 }
  218.             ]
  219.         }
  220.     }
  221.     dataxTemplate['job']['setting']['speed']['channel'] = channel
  222.     dataxTemplateContent = dataxTemplate['job']['content'][0]
  224.     pluginName = ''
  225.     if paramsDict.get('datasourceType'):
  226.         pluginName = '%s%s' % (paramsDict['datasourceType'], readerOrWriter)
  227.     elif paramsDict.get('jdbcUrl'):
  228.         pluginName = parsePluginName(paramsDict['jdbcUrl'], readerOrWriter)
  229.     elif paramsDict.get('url'):
  230.         pluginName = 'adswriter'
  232.     theOtherSide = 'writer' if readerOrWriter == 'reader' else 'reader'
  233.     dataxPluginParamsContent = dataxTemplateContent.get(readerOrWriter).get('parameter')
  234.     dataxPluginParamsContent.update(paramsDict)
  236.     dataxPluginParamsContentOtherSide = dataxTemplateContent.get(theOtherSide).get('parameter')
  238.     if readerOrWriter == 'reader':
  239.         dataxTemplateContent.get('reader')['name'] = pluginName
  240.         dataxTemplateContent.get('writer')['name'] = 'streamwriter'
  241.         if paramsDict.get('writer-print'):
  242.             dataxPluginParamsContentOtherSide['print'] = paramsDict['writer-print']
  243.             del dataxPluginParamsContent['writer-print']
  244.         del dataxPluginParamsContentOtherSide['connection']
  245.     if readerOrWriter == 'writer':
  246.         dataxTemplateContent.get('reader')['name'] = 'streamreader'
  247.         dataxTemplateContent.get('writer')['name'] = pluginName
  248.         if paramsDict.get('reader-column'):
  249.             dataxPluginParamsContentOtherSide['column'] = paramsDict['reader-column']
  250.             del dataxPluginParamsContent['reader-column']
  251.         if paramsDict.get('reader-sliceRecordCount'):
  252.             dataxPluginParamsContentOtherSide['sliceRecordCount'] = paramsDict['reader-sliceRecordCount']
  253.             del dataxPluginParamsContent['reader-sliceRecordCount']
  254.         del dataxPluginParamsContentOtherSide['connection']
  256.     if paramsDict.get('jdbcUrl'):
  257.         if readerOrWriter == 'reader':
  258.             dataxPluginParamsContent['connection'][0]['jdbcUrl'].append(paramsDict['jdbcUrl'])
  259.         else:
  260.             dataxPluginParamsContent['connection'][0]['jdbcUrl'] = paramsDict['jdbcUrl']
  261.     if paramsDict.get('table'):
  262.         dataxPluginParamsContent['connection'][0]['table'].append(paramsDict['table'])
  265.     traceJobJson = json.dumps(dataxTemplate, indent = 4)
  266.     return traceJobJson
  268. def isUrl(path):
  269.     if not path:
  270.         return False
  271.     if not isinstance(path, str):
  272.         raise Exception('Configuration file path required for the string, you configure is:%s' % path)
  273.     m = re.match(r"^http[s]?://\S+\w*", path.lower())
  274.     if m:
  275.         return True
  276.     else:
  277.         return False
  280. def readJobJsonFromLocal(jobConfigPath):
  281.     jobConfigContent = None
  282.     jobConfigPath = os.path.abspath(jobConfigPath)
  283.     file = open(jobConfigPath)
  284.     try:
  285.         jobConfigContent = file.read()
  286.     finally:
  287.         file.close()
  288.     if not jobConfigContent:
  289.         raise Exception("Your job configuration file read the result is empty, please check the configuration is legal, path: [%s]\nconfiguration:\n%s" % (jobConfigPath, str(jobConfigContent)))
  290.     return jobConfigContent
  293. def readJobJsonFromRemote(jobConfigPath):
  294.     import urllib
  295.     conn = urllib.urlopen(jobConfigPath)
  296.     jobJson = conn.read()
  297.     return jobJson
  299. def parseJson(strConfig, context):
  300.     try:
  301.         return json.loads(strConfig)
  302.     except Exception, e:
  303.         import traceback
  304.         traceback.print_exc()
  305.         sys.stdout.flush()
  306.         print >> sys.stderr, '%s %s need in line with json syntax' % (context, strConfig)
  307.         sys.exit(-1)
  309. def convert(options, args):
  310.     traceJobJson = ''
  311.     if options.file:
  312.         if isUrl(options.file):
  313.             traceJobJson = readJobJsonFromRemote(options.file)
  314.         else:
  315.             traceJobJson = readJobJsonFromLocal(options.file)
  316.         traceJobDict = parseJson(traceJobJson, '%s content' % options.file)
  317.         attributeNotNone(traceJobDict, ['job'])
  318.         attributeNotNone(traceJobDict['job'], ['content'])
  319.         attributeNotNone(traceJobDict['job']['content'][0], ['reader', 'writer'])
  320.         attributeNotNone(traceJobDict['job']['content'][0]['reader'], ['name', 'parameter'])
  321.         attributeNotNone(traceJobDict['job']['content'][0]['writer'], ['name', 'parameter'])
  322.         if options.type == 'reader':
  323.             traceJobDict['job']['content'][0]['writer']['name'] = 'streamwriter'
  324.             if options.reader:
  325.                 traceReaderDict = parseJson(options.reader, 'reader config')
  326.                 if traceReaderDict.get('writer-print') is not None:
  327.                     traceJobDict['job']['content'][0]['writer']['parameter']['print'] = traceReaderDict.get('writer-print')
  328.                 else:
  329.                     traceJobDict['job']['content'][0]['writer']['parameter']['print'] = 'false'
  330.             else:
  331.                 traceJobDict['job']['content'][0]['writer']['parameter']['print'] = 'false'
  332.         elif options.type == 'writer':
  333.             traceJobDict['job']['content'][0]['reader']['name'] = 'streamreader'
  334.             if options.writer:
  335.                 traceWriterDict = parseJson(options.writer, 'writer config')
  336.                 if traceWriterDict.get('reader-column'):
  337.                     traceJobDict['job']['content'][0]['reader']['parameter']['column'] = traceWriterDict['reader-column']
  338.                 if traceWriterDict.get('reader-sliceRecordCount'):
  339.                     traceJobDict['job']['content'][0]['reader']['parameter']['sliceRecordCount'] = traceWriterDict['reader-sliceRecordCount']
  340.             else:
  341.                 columnSize = len(traceJobDict['job']['content'][0]['writer']['parameter']['column'])
  342.                 streamReaderColumn = []
  343.                 for i in range(columnSize):
  344.                     streamReaderColumn.append({"type": "long", "random": "2,10"})
  345.                 traceJobDict['job']['content'][0]['reader']['parameter']['column'] = streamReaderColumn
  346.                 traceJobDict['job']['content'][0]['reader']['parameter']['sliceRecordCount'] = 10000
  347.         else:
  348.             pass#do nothing
  349.         return json.dumps(traceJobDict, indent = 4)
  350.     elif options.reader:
  351.         traceReaderDict = parseJson(options.reader, 'reader config')
  352.         return renderDataXJson(traceReaderDict, 'reader', options.channel)
  353.     elif options.writer:
  354.         traceWriterDict = parseJson(options.writer, 'writer config')
  355.         return renderDataXJson(traceWriterDict, 'writer', options.channel)
  356.     else:
  357.         print getUsage()
  358.         sys.exit(-1)
  359.     #dataxParams = {}
  360.     #for opt, value in options.__dict__.items():
  361.     #    dataxParams[opt] = value
  362. ##end datax json generate logic
  365. if __name__ == "__main__":
  366.     printCopyright()
  367.     parser = getOptionParser()
  369.     options, args = parser.parse_args(sys.argv[1:])
  370.     #print options, args
  371.     dataxTraceJobJson = convert(options, args)
  373.     #鐢盡AC鍦板潃銆佸綋鍓嶆椂闂存埑銆侀殢鏈烘暟鐢熸垚,鍙互淇濊瘉鍏ㄧ悆鑼冨洿鍐呯殑鍞竴鎬?    dataxJobPath = os.path.join(os.getcwd(), "perftrace-" + str(uuid.uuid1()))
  374.     jobConfigOk = True
  375.     if os.path.exists(dataxJobPath):
  376.         print "file already exists, truncate and rewrite it? %s" % dataxJobPath
  377.         if yesNoChoice():
  378.             jobConfigOk = True
  379.         else:
  380.             print "exit failed, because of file conflict"
  381.             sys.exit(-1)
  382.     fileWriter = open(dataxJobPath, 'w')
  383.     fileWriter.write(dataxTraceJobJson)
  384.     fileWriter.close()
  387.     print "trace environments:"
  388.     print "dataxJobPath:  %s" % dataxJobPath
  389.     dataxHomePath = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
  390.     print "dataxHomePath: %s" % dataxHomePath
  392.     dataxCommand = "%s %s" % (os.path.join(dataxHomePath, "bin", "datax.py"), dataxJobPath)
  393.     print "dataxCommand:  %s" % dataxCommand
  395.     returncode = fork(dataxCommand, True)
  396.     if options.delete == 'true':
  397.         os.remove(dataxJobPath)
  398.     sys.exit(returncode)

DataX 基本使用

查看模板

查看 streamreader —> streamwriter 的模板

  1. python3 /usr/local/datax/bin/datax.py -r streamreader -w streamwriter

输出

  1. DataX (DATAX-OPENSOURCE-3.0), From Alibaba !
  2. Copyright (C) 2010-2017, Alibaba Group. All Rights Reserved.
  5. Please refer to the streamreader document:
  6.      https://github.com/alibaba/DataX/blob/master/streamreader/doc/streamreader.md
  8. Please refer to the streamwriter document:
  9.      https://github.com/alibaba/DataX/blob/master/streamwriter/doc/streamwriter.md
  11. Please save the following configuration as a json file and  use
  12.      python {DATAX_HOME}/bin/datax.py {JSON_FILE_NAME}.json
  13. to run the job.
  15. {
  16.     "job": {
  17.         "content": [
  18.             {
  19.                 "reader": {
  20.                     "name": "streamreader",
  21.                     "parameter": {
  22.                         "column": [],
  23.                         "sliceRecordCount": ""
  24.                     }
  25.                 },
  26.                 "writer": {
  27.                     "name": "streamwriter",
  28.                     "parameter": {
  29.                         "encoding": "",
  30.                         "print": true
  31.                     }
  32.                 }
  33.             }
  34.         ],
  35.         "setting": {
  36.             "speed": {
  37.                 "channel": ""
  38.             }
  39.         }
  40.     }
  41. }

run test job

根据模板编写 json 文件

  1. cd /usr/local/datax/job
  1. cat <<END > test.json
  2. {
  3.     "job": {
  4.         "content": [
  5.             {
  6.                 "reader": {
  7.                     "name": "streamreader", 
  8.                     "parameter": {
  9.                         "column": [                             # 同步的列名 (* 表示所有)
  10.                 {
  11.                     "type":"string",
  12.                 "value":"Hello."
  13.                 },
  14.                 {
  15.                     "type":"string",
  16.                 "value":"河北彭于晏"
  17.                 },
  18.             ], 
  19.                         "sliceRecordCount": "3"                 # 打印数量
  20.                     }
  21.                 }, 
  22.                 "writer": {
  23.                     "name": "streamwriter", 
  24.                     "parameter": {
  25.                         "encoding": "utf-8",                    # 编码
  26.                         "print": true
  27.                     }
  28.                 }
  29.             }
  30.         ], 
  31.         "setting": {
  32.             "speed": {
  33.                 "channel": "2"                                  # 并发 (即 sliceRecordCount * channel = 结果)
  34.             }
  35.         }
  36.     }
  37. }

输出：（要是复制上面的话，需要把 # 带的内容去掉）

  1. cat <<END > test.json
  2. {
  3.     "job": {
  4.         "content": [
  5.             {
  6.                 "reader": {
  7.                     "name": "streamreader",
  8.                     "parameter": {
  9.                         "column": [
  10.                             {
  11.                                 "type": "string",
  12.                                 "value": "Hello."
  13.                             },
  14.                             {
  15.                                 "type": "string",
  16.                                 "value": "河北彭于晏"
  17.                             },
  18.                         ],
  19.                         "sliceRecordCount": "3"
  20.                     }
  21.                 },
  22.                 "writer": {
  23.                     "name": "streamwriter",
  24.                     "parameter": {
  25.                         "encoding": "utf-8",
  26.                         "print": true
  27.                     }
  28.                 }
  29.             }
  30.         ],
  31.         "setting": {
  32.             "speed": {
  33.                 "channel": "2"
  34.             }
  35.         }
  36.     }
  37. }

再在命令行输入END,验证test.json

  1. python3 /usr/local/datax/bin/datax.py test.json

输出
1665238434815.png