O Alibaba DataX foi originalmente desenvolvido com base no Python 2.7, o que gera incompatibilidades sintáticas ao tentar executá-lo em ambientes modernos que utilizam o Python 3. Para contornar essa limitação sem recorrer a ambientes virtuais legados ou soluções pagas, é necessário atualizar os scripts de inicialização localizados no diretório bin da instalação.
Antes de proceder, faça um backup dos arquivos originasi em datax/bin/. Em seguida, substitua os três scripts abaixo pelas versões adaptadas para a sintaxe do Python 3.
- Substituição do arquivo
datax.py
Este script é responsável pela orquestarção do processo de execução. A versão abaixo corrige as instruções de impressão, manipulação de strings e chamadas de sistema para o padrão atual.
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import sys
import os
import signal
import subprocess
import time
import re
import socket
import json
from optparse import OptionParser, OptionGroup
from string import Template
import codecs
import platform
def is_windows_env():
return platform.system() == 'Windows'
ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
APP_VERSION = 'DATAX-OPENSOURCE-3.0'
if is_windows_env():
codecs.register(lambda name: codecs.lookup('utf-8') if name == 'cp65001' else None)
LIB_PATH = f"{ROOT_DIR}/lib/*"
else:
LIB_PATH = f"{ROOT_DIR}/lib/*:."
LOG_CONFIG = f"{ROOT_DIR}/conf/logback.xml"
BASE_JVM_OPTS = "-Xms1g -Xmx1g -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=%s/log" % ROOT_DIR
BASE_JAVA_OPTS = "-Dfile.encoding=UTF-8 -Dlogback.statusListenerClass=ch.qos.logback.core.status.NopStatusListener -Djava.security.egd=file:///dev/urandom -Ddatax.home=%s -Dlogback.configurationFile=%s" % (ROOT_DIR, LOG_CONFIG)
ENGINE_CMD_TEMPLATE = "java -server ${jvm} %s -classpath %s ${params} com.alibaba.datax.core.Engine -mode ${mode} -jobid ${jobid} -job ${job}" % (BASE_JAVA_OPTS, LIB_PATH)
DEBUG_OPTS = "-Xdebug -Xrunjdwp:transport=dt_socket,server=y,address=9999"
EXIT_CODES = {"KILL": 143, "FAIL": -1, "OK": 0, "RUN": 1, "RETRY": 2}
proc_child = None
def resolve_local_ip():
try:
return socket.gethostbyname(socket.getfqdn(socket.gethostname()))
except Exception:
return "Desconhecido"
def handle_signal(sig_num, frame):
global proc_child
print(f"[Erro] DataX recebeu sinal inesperado {sig_num}, encerrando processo.", file=sys.stderr)
if proc_child:
proc_child.send_signal(signal.SIGQUIT)
time.sleep(1)
proc_child.kill()
print("Processo do DataX foi finalizado!", file=sys.stderr)
sys.exit(EXIT_CODES["KILL"])
def setup_signals():
if not is_windows_env():
global proc_child
signal.signal(signal.SIGINT, handle_signal)
signal.signal(signal.SIGQUIT, handle_signal)
signal.signal(signal.SIGTERM, handle_signal)
def create_parser():
parser = OptionParser(usage="uso: %prog [opções] caminho-ou-url-do-job")
prod_group = OptionGroup(parser, "Opções de Produção", "Parâmetros para ambiente produtivo.")
prod_group.add_option("-j", "--jvm", dest="jvm_args", default=BASE_JVM_OPTS, help="Personalizações de JVM.")
prod_group.add_option("--jobid", dest="job_id", default="-1", help="Identificador único do job.")
prod_group.add_option("-m", "--mode", dest="run_mode", default="standalone", help="Modo de execução: standalone, local, distribute.")
prod_group.add_option("-p", "--params", dest="job_params", help="Parâmetros dinâmicos do job, ex: -p\"-DtableName=tbl\".")
prod_group.add_option("-r", "--reader", dest="reader_plugin", help="Consulta modelo do reader (ex: mysqlreader).")
prod_group.add_option("-w", "--writer", dest="writer_plugin", help="Consulta modelo do writer (ex: mysqlwriter).")
parser.add_option_group(prod_group)
dev_group = OptionGroup(parser, "Opções de Desenvolvimento", "Ferramentas para debug e análise.")
dev_group.add_option("-d", "--debug", dest="enable_debug", action="store_true", help="Ativa modo de debug remoto.")
dev_group.add_option("--loglevel", dest="log_level", default="info", help="Nível de log: debug, info, all.")
parser.add_option_group(dev_group)
return parser
def load_plugin_template(plugin_path):
with open(plugin_path, 'r', encoding='utf-8') as f:
return json.load(f)
def generate_config_template(reader_name, writer_name):
doc_url = "https://github.com/alibaba/DataX/blob/master/{}/doc/{}.md"
print(f"Referência do Reader: {doc_url.format(reader_name, reader_name)}")
print(f"Referência do Writer: {doc_url.format(writer_name, writer_name)}")
print("Salve a configuração abaixo como um arquivo JSON e execute o job.")
base_template = {"job": {"setting": {"speed": {"channel": ""}}, "content": [{"reader": {}, "writer": {}}]}}
try:
reader_cfg = load_plugin_template(f"{ROOT_DIR}/plugin/reader/{reader_name}/plugin_job_template.json")
base_template['job']['content'][0]['reader'] = reader_cfg
except Exception:
print(f"Erro ao carregar template do reader: {reader_name}")
try:
writer_cfg = load_plugin_template(f"{ROOT_DIR}/plugin/writer/{writer_name}/plugin_job_template.json")
base_template['job']['content'][0]['writer'] = writer_cfg
except Exception:
print(f"Erro ao carregar template do writer: {writer_name}")
print(json.dumps(base_template, indent=4, sort_keys=True))
def is_http_url(path):
if not isinstance(path, str):
return False
return bool(re.match(r"^http[s]?://\S+\w*", path.lower()))
def assemble_command(cli_opts, job_args):
cmd_vars = {}
current_jvm = BASE_JVM_OPTS
if cli_opts.jvm_args:
current_jvm += f" {cli_opts.jvm_args}"
if cli_opts.enable_debug:
current_jvm += f" {DEBUG_OPTS}"
print(f"IP local: {resolve_local_ip()}")
if cli_opts.log_level:
current_jvm += f" -Dloglevel={cli_opts.log_level}"
if cli_opts.run_mode:
cmd_vars["mode"] = cli_opts.run_mode
job_source = job_args[0]
if not is_http_url(job_source):
job_source = os.path.abspath(job_source)
if job_source.lower().startswith("file://"):
job_source = job_source[7:]
log_name = job_source[-20:].replace('/', '_').replace('.', '_')
job_params = f"-Dlog.file.name={log_name}"
if cli_opts.job_params:
job_params += f" {cli_opts.job_params}"
if cli_opts.job_id:
cmd_vars["jobid"] = cli_opts.job_id
cmd_vars["jvm"] = current_jvm
cmd_vars["params"] = job_params
cmd_vars["job"] = job_source
return Template(ENGINE_CMD_TEMPLATE).substitute(**cmd_vars)
def show_banner():
print(f"""
DataX ({APP_VERSION}), Desenvolvido pela Alibaba!
Copyright (C) 2010-2017, Alibaba Group. Todos os direitos reservados.
""")
sys.stdout.flush()
if __name__ == "__main__":
show_banner()
parser = create_parser()
opts, args = parser.parse_args(sys.argv[1:])
if opts.reader_plugin and opts.writer_plugin:
generate_config_template(opts.reader_plugin, opts.writer_plugin)
sys.exit(EXIT_CODES['OK'])
if len(args) != 1:
parser.print_help()
sys.exit(EXIT_CODES['FAIL'])
exec_cmd = assemble_command(opts, args)
global proc_child
proc_child = subprocess.Popen(exec_cmd, shell=True)
setup_signals()
proc_child.wait()
sys.exit(proc_child.returncode)
- Subtsituição do arquivo
dxprof.py
Este utilitário realiza a análise de logs para extração de métricas de performance por tarefa. As regexes e manipulações de tempo foram ajustadas para Python 3.
#!/usr/bin/env python3
import re
import sys
import time
PATTERN_SQL_START = re.compile(r'Begin\s+to\s+read\s+record\s+by\s+Sql', re.IGNORECASE)
PATTERN_SQL_END = re.compile(r'Finished\s+read\s+record\s+by\s+Sql', re.IGNORECASE)
PATTERN_TABLE_NAME = re.compile(r'from\s+(\w+)(\s+where|\s*$)', re.IGNORECASE)
PATTERN_JDBC_CONN = re.compile(r'jdbcUrl:\s*\[(.+?)\]', re.IGNORECASE)
PATTERN_TASK_UUID = re.compile(r'(\d+\-)+reader')
PATTERN_COMMIT_UUID = re.compile(r'(\d+\-)+writer')
PATTERN_COMMIT_START = re.compile(r'begin\s+to\s+commit\s+blocks', re.IGNORECASE)
PATTERN_COMMIT_END = re.compile(r'commit\s+blocks\s+ok', re.IGNORECASE)
def extract_timestamp(log_line):
try:
return int(time.mktime(time.strptime(log_line[:19], '%Y-%m-%d %H:%M:%S')))
except Exception:
return 0
def extract_host(log_line):
match = PATTERN_JDBC_CONN.search(log_line)
if not match:
return ''
conn_str = match.group(1).split('?')[0]
at_pos = conn_str.find('@')
if at_pos > -1:
return conn_str[at_pos+1:].lower()
slash_pos = conn_str.find('//')
if slash_pos > -1:
return conn_str[slash_pos+2:].lower()
return ''
def extract_table(log_line):
match = PATTERN_TABLE_NAME.search(log_line)
return match.group(1).lower() if match else ''
def analyze_log_file(target_file):
global last_reader_uuid, commit_map, reader_map, epoch
last_reader_uuid = ''
reader_map = {}
commit_map = {}
epoch = int(time.time())
with open(target_file, 'r', encoding='utf-8') as fh:
for raw_line in fh:
line = raw_line.strip()
if last_reader_uuid and last_reader_uuid in reader_map:
reader_map[last_reader_uuid]['host'] = extract_host(line)
last_reader_uuid = ''
if 'CommonRdbmsReader$Task' in line:
process_read_task(line)
elif 'commit blocks' in line:
process_commit_task(line)
def process_read_task(line):
global last_reader_uuid
uuid_match = PATTERN_TASK_UUID.search(line)
if not uuid_match:
return
last_reader_uuid = uuid_match.group()
if PATTERN_SQL_START.search(line):
reader_map[last_reader_uuid] = {
'status': 'R', 'start': extract_timestamp(line),
'end': epoch, 'host': extract_host(line), 'table': extract_table(line)
}
elif last_reader_uuid in reader_map and PATTERN_SQL_END.search(line):
reader_map[last_reader_uuid]['status'] = 'D'
reader_map[last_reader_uuid]['end'] = extract_timestamp(line)
def process_commit_task(line):
global last_commit_uuid
uuid_match = PATTERN_COMMIT_UUID.search(line)
if not uuid_match:
return
last_commit_uuid = uuid_match.group()
if PATTERN_COMMIT_START.search(line):
commit_map[last_commit_uuid] = {
'status': 'R', 'start': extract_timestamp(line), 'end': epoch
}
elif last_commit_uuid in commit_map and PATTERN_COMMIT_END.search(line):
commit_map[last_commit_uuid]['status'] = 'D'
commit_map[last_commit_uuid]['end'] = extract_timestamp(line)
def print_metrics():
def duration_sorter(a, b):
return b['duration'] - a['duration']
tasks = []
servers = set()
stats = {'total_time': 0, 'count': 0, 'max_end': 0, 'min_start': epoch}
commit_tasks = []
commit_stats = {'total_time': 0, 'count': 0}
for uid, data in reader_map.items():
data['id'] = uid
data['duration'] = data['end'] - data['start']
tasks.append(data)
if data['host']:
servers.add(data['host'])
if 0 < data['duration'] < 864000:
stats['total_time'] += data['duration']
stats['count'] += 1
stats['max_end'] = max(stats['max_end'], data['end'])
stats['min_start'] = min(stats['min_start'], data['start'])
for uid, data in commit_map.items():
data['id'] = uid
data['duration'] = data['end'] - data['start']
commit_tasks.append(data)
if 0 < data['duration'] < 864000:
commit_stats['total_time'] += data['duration']
commit_stats['count'] += 1
total_window = max(1, stats['max_end'] - stats['min_start'])
throughput = stats['count'] / (stats['total_time'] or total_window)
tasks.sort(duration_sorter)
for t in tasks:
end_str = time.strftime('%H:%M:%S', time.localtime(t['end'])) if t['status'] == 'D' else '--'
print(f"{t['status']}\t{t['host']}.{t['table']}\t{time.strftime('%H:%M:%S', time.localtime(t['start']))}\t{end_str}\t{t['duration']:4d}\t{100*t['duration']/total_window:.1f}%\t{throughput*t['duration']:.2f}")
if not tasks or not stats['count']:
return
print('\n--- Estatísticas de Perfilamento DataX ---')
print(f"{stats['count']} tarefa(s) em {len(servers)} servidor(es), Duração total {stats['total_time']}s, Média {stats['total_time']/stats['count']:.2f}s por tarefa")
print(f"Custo real {total_window}s ({time.strftime('%H:%M:%S', time.localtime(stats['min_start']))} - {time.strftime('%H:%M:%S', time.localtime(stats['max_end']))}), Concorrência: {stats['total_time']/total_window:.2f}, Índice de desvio: {throughput*tasks[0]['duration']:.2f}")
commit_throughput = commit_stats['count'] / (commit_stats['total_time'] or total_window)
commit_tasks.sort(duration_sorter)
if commit_tasks:
print(f"{commit_stats['count']} tarefa(s) commit, Total {commit_stats['total_time']}s, Média {commit_stats['total_time']/commit_stats['count']:.2f}s, Desvio: {commit_throughput*commit_tasks[0]['duration']:.2f}")
if len(sys.argv) < 2:
print(f"Uso: {sys.argv[0]} <arquivo_de_log>")
sys.exit(1)
else:
analyze_log_file(sys.argv[1])
print_metrics()
- Substituição do arquivo
perftrace.py
Responsável pela geração de Jobs de teste e medição de throughput. A versão atualizada remove dependências legadas do Python 2, ajusta a manipulação de encoding e moderniza as chamadas de rede e sistema.
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import re
import os
import sys
import json
import uuid
import signal
import time
import subprocess
import urllib.request
from optparse import OptionParser
def build_parser():
parser = OptionParser(usage=get_help_text())
parser.add_option('-r', '--reader', dest='reader_cfg', help='Configuração JSON para teste de leitura.')
parser.add_option('-w', '--writer', dest='writer_cfg', help='Configuração JSON para teste de escrita.')
parser.add_option('-c', '--channel', dest='channels', default='1', help='Número de threads concorrentes (padrão: 1).')
parser.add_option('-f', '--file', dest='config_file', help='Arquivo JSON de configuração existente.')
parser.add_option('-t', '--type', dest='trace_type', default='reader', help='Tipo de teste: reader ou writer.')
parser.add_option('-d', '--delete', dest='clean_temp', default='true', help='Remove arquivos temporários após execução.')
return parser
def get_help_text():
return '''
Parâmetros disponíveis para -r --reader:
datasourceType: tipo de fonte (mysql, oracle, postgresql, etc.)
jdbcUrl: string de conexão JDBC
username / password: credenciais de acesso
table: nome da tabela alvo
column: colunas a serem lidas (padrão: ['*'])
splitPk: coluna de particionamento
where: filtro opcional
fetchSize: lote de fetch
reader-sliceRecordCount: quantidade de dados mockados
reader-column: definição de colunas mockadas
Parâmetros disponíveis para -w --writer:
datasourceType, jdbcUrl, username, password, table, column
batchSize: lote de escrita (padrão: 512)
preSql / postSql: comandos SQL de preparação e finalização
writer-print: habilita impressão de dados (padrão: false)
Controle global:
-c --channel: concorrência
-f --file: caminho para JSON completo
-t --type: reader ou writer
Exemplo: perftrace.py --channel=10 --reader='{"jdbcUrl":"jdbc:mysql://localhost:3306/db", "table":"test", "username":"u", "password":"p"}'
'''
def show_banner():
print("""
DataX Util Tools, Desenvolvido pela Alibaba!
Copyright (C) 2010-2016, Alibaba Group. Todos os direitos reservados.""")
sys.stdout.flush()
def ask_confirmation():
resp = input("Confirma? (s/n): ").strip().lower()
return resp in ('yes', 'y', 'ye', 's', '')
child_proc = None
def handle_sig(sig, frame):
global child_proc
print(f"[Erro] Sinal {sig} recebido, encerrando processo.", file=sys.stderr)
if child_proc:
child_proc.send_signal(signal.SIGQUIT)
time.sleep(1)
child_proc.kill()
print("Processo finalizado!", file=sys.stderr)
sys.exit(-1)
def setup_sig():
global child_proc
signal.signal(signal.SIGINT, handle_sig)
signal.signal(signal.SIGQUIT, handle_sig)
signal.signal(signal.SIGTERM, handle_sig)
def run_subprocess(cmd, shell=True):
global child_proc
child_proc = subprocess.Popen(cmd, shell=shell)
setup_sig()
child_proc.wait()
return child_proc.returncode
def validate_not_empty(value, context):
if not value:
raise ValueError(f"Propriedade obrigatória vazia: {context}")
def validate_keys(obj, keys):
for k in keys:
validate_not_empty(obj.get(k), k)
def detect_plugin(jdbc, suffix):
if re.search(r'jdbc:mysql://', jdbc): return f"mysql{suffix}"
if re.search(r'jdbc:postgresql://', jdbc): return f"postgresql{suffix}"
if re.search(r'jdbc:oracle:', jdbc): return f"oracle{suffix}"
if re.search(r'jdbc:sqlserver://', jdbc): return f"sqlserver{suffix}"
if re.search(r'jdbc:db2://', jdbc): return f"db2{suffix}"
return f"plugin{suffix}"
def build_trace_json(cfg_dict, side='reader', chan=1):
tpl = {
"job": {
"setting": {"speed": {"channel": chan}},
"content": [{
"reader": {"name": "", "parameter": {"username": "", "password": "", "sliceRecordCount": "10000", "column": ["*"], "connection": [{"table": [], "jdbcUrl": []}]}},
"writer": {"name": "", "parameter": {"print": "false", "connection": [{"table": [], "jdbcUrl": ""}]}}
}]
}
}
content = tpl['job']['content'][0]
plug_name = cfg_dict.get('datasourceType', '') + side if cfg_dict.get('datasourceType') else detect_plugin(cfg_dict.get('jdbcUrl', ''), side)
if cfg_dict.get('url'): plug_name = 'adswriter'
other_side = 'writer' if side == 'reader' else 'reader'
current_params = content[side]['parameter']
current_params.update(cfg_dict)
other_params = content[other_side]['parameter']
if side == 'reader':
content['reader']['name'] = plug_name
content['writer']['name'] = 'streamwriter'
if 'writer-print' in cfg_dict:
other_params['print'] = cfg_dict.pop('writer-print')
other_params.pop('connection', None)
else:
content['reader']['name'] = 'streamreader'
content['writer']['name'] = plug_name
if 'reader-column' in cfg_dict:
other_params['column'] = cfg_dict.pop('reader-column')
if 'reader-sliceRecordCount' in cfg_dict:
other_params['sliceRecordCount'] = cfg_dict.pop('reader-sliceRecordCount')
other_params.pop('connection', None)
if cfg_dict.get('jdbcUrl'):
if side == 'reader':
current_params['connection'][0]['jdbcUrl'].append(cfg_dict['jdbcUrl'])
else:
current_params['connection'][0]['jdbcUrl'] = cfg_dict['jdbcUrl']
if cfg_dict.get('table'):
current_params['connection'][0]['table'].append(cfg_dict['table'])
return json.dumps(tpl, indent=4)
def load_json_remote(url):
with urllib.request.urlopen(url) as res:
return res.read().decode('utf-8')
def load_json_local(path):
abs_path = os.path.abspath(path)
with open(abs_path, 'r', encoding='utf-8') as f:
data = f.read()
if not data:
raise ValueError(f"Arquivo vazio ou ilegível: {abs_path}")
return data
def parse_json_str(raw, ctx):
try:
return json.loads(raw)
except Exception as e:
print(f"Erro de sintaxe JSON em {ctx}: {e}", file=sys.stderr)
sys.exit(-1)
def prepare_config(opts, args):
if opts.config_file:
raw = load_json_remote(opts.config_file) if opts.config_file.startswith('http') else load_json_local(opts.config_file)
job = parse_json_str(raw, 'arquivo de configuração')
validate_keys(job, ['job'])
validate_keys(job['job'], ['content'])
first = job['job']['content'][0]
validate_keys(first, ['reader', 'writer'])
validate_keys(first['reader'], ['name', 'parameter'])
validate_keys(first['writer'], ['name', 'parameter'])
if opts.trace_type == 'reader':
first['writer']['name'] = 'streamwriter'
if opts.reader_cfg:
r_cfg = parse_json_str(opts.reader_cfg, 'reader')
first['writer']['parameter']['print'] = r_cfg.get('writer-print', 'false')
else:
first['writer']['parameter']['print'] = 'false'
elif opts.trace_type == 'writer':
first['reader']['name'] = 'streamreader'
if opts.writer_cfg:
w_cfg = parse_json_str(opts.writer_cfg, 'writer')
if 'reader-column' in w_cfg: first['reader']['parameter']['column'] = w_cfg['reader-column']
if 'reader-sliceRecordCount' in w_cfg: first['reader']['parameter']['sliceRecordCount'] = w_cfg['reader-sliceRecordCount']
else:
col_count = len(first['writer']['parameter'].get('column', ['*']))
first['reader']['parameter']['column'] = [{"type": "long", "random": "2,10"} for _ in range(col_count)]
first['reader']['parameter']['sliceRecordCount'] = 10000
return json.dumps(job, indent=4)
elif opts.reader_cfg:
return build_trace_json(parse_json_str(opts.reader_cfg, 'reader'), 'reader', int(opts.channels))
elif opts.writer_cfg:
return build_trace_json(parse_json_str(opts.writer_cfg, 'writer'), 'writer', int(opts.channels))
else:
print(get_help_text())
sys.exit(-1)
if __name__ == "__main__":
show_banner()
parser = build_parser()
opts, args = parser.parse_args(sys.argv[1:])
final_json = prepare_config(opts, args)
tmp_path = os.path.join(os.getcwd(), f"perftrace-{uuid.uuid1()}")
if os.path.exists(tmp_path):
if not ask_confirmation():
print("Cancelado devido a conflito de arquivo.")
sys.exit(-1)
with open(tmp_path, 'w', encoding='utf-8') as f:
f.write(final_json)
print("Ambiente de teste preparado:")
print(f"Caminho do job: {tmp_path}")
base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
print(f"Diretório base: {base_dir}")
cmd = f"{os.path.join(base_dir, 'bin', 'datax.py')} {tmp_path}"
print(f"Comando: {cmd}")
exit_code = run_subprocess(cmd)
if opts.clean_temp == 'true':
os.remove(tmp_path)
sys.exit(exit_code)