datax增量更新修改python启动脚本
2021-10-19 本文已影响0人
yichen_china
完整json文件如下
注意俩参数
job.difMinutes=10分
job.minDifMinutes=5分
#意思是每次脚本执行的查询时间间隔10分。
替换时间 (?) 会被替换成上次执行时间 +job.difMinutes 时间
job.content.reader.where": "update_time between (?)"
最低间隔时间，执行时间最低5分钟，低于5分钟不执行
job.minDifMinutes=5分
完整配置json如下
datax/bin/beiyao/stream.json
{
"job": {
"minDifMinutes":5,
"difMinutes":10,
"content": [
{
"writer": {
"parameter": {
"username": "beiyao",
"writeMode": "update",
"column": [
"id",
"goods_no",
"store_no",
"stock",
"sku_id",
"price",
"cost_price",
"source_create_time",
"source_type",
"source_update_time"
],
"splitPk": "id",
"connection": [
{
"table": [
"zt_store_stock"
],
"jdbcUrl": "jdbc:mysql://192.168.3.183:3306/beiyao"
}
],
"password": "wHm6HRNfBDYAdcjd"
},
"name": "mysqlwriter"
},
"reader": {
"parameter": {
"username": "mg",
"encoding": "UTF-8",
"column": [
"Id",
"goods_no",
"store_no",
"stock",
"Code",
"price",
"cost_price",
"update_time as create_time",
"'tc' as source_type",
"update_time"
],
"splitPk": "Id",
"connection": [
{
"table": [
"MG_DRUGSTORE"
],
"jdbcUrl": [
"jdbc:sqlserver://192.168.2.222:1433;DatabaseName=bylsn"
]
}
],
"password": "mg",
"where": "update_time between (?)"
},
"name": "sqlserverreader"
}
}
],
"setting": {
"speed": {
"byte": 1048576,
"channel": 3
},
"errorLimit": {
"record": 100,
"percentage": 0.5
}
}
}
}
缓存时间json如下
根目录
datax/cache.json
key值是 json配置文件的目录名+文件全名去斜杠
{"beiyaostream.json": "2021-10-18 19:07:35"}
datax/bin 目录
datax.py 脚本如下
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import sys
import os
import signal
import subprocess
import time
import re
import socket
import json
from optparse import OptionParser
from optparse import OptionGroup
from string import Template
import codecs
import platform
import datetime
import io
NOWTIME=datetime.datetime.now()
def isWindows():
return platform.system() == 'Windows'
DATAX_HOME = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DATAX_VERSION = 'DATAX-OPENSOURCE-3.0'
if isWindows():
codecs.register(lambda name: name == 'cp65001' and codecs.lookup('utf-8') or None)
CLASS_PATH = ("%s/lib/*") % (DATAX_HOME)
else:
CLASS_PATH = ("%s/lib/*:.") % (DATAX_HOME)
LOGBACK_FILE = ("%s/conf/logback.xml") % (DATAX_HOME)
DEFAULT_JVM = "-Xms1g -Xmx1g -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=%s/log" % (DATAX_HOME)
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" % (
DATAX_HOME, LOGBACK_FILE)
ENGINE_COMMAND = "java -server ${jvm} %s -classpath %s ${params} com.alibaba.datax.core.Engine -mode ${mode} -jobid ${jobid} -job ${job}" % (
DEFAULT_PROPERTY_CONF, CLASS_PATH)
REMOTE_DEBUG_CONFIG = "-Xdebug -Xrunjdwp:transport=dt_socket,server=y,address=9999"
RET_STATE = {
"KILL": 143,
"FAIL": -1,
"OK": 0,
"RUN": 1,
"RETRY": 2
}
CACHEFile=DATAX_HOME+'\cache.json'
def getLocalIp():
try:
return socket.gethostbyname(socket.getfqdn(socket.gethostname()))
except:
return "Unknown"
def suicide(signum, e):
global child_process
print >> sys.stderr, "[Error] DataX receive unexpected signal %d, starts to suicide." % (signum)
if child_process:
child_process.send_signal(signal.SIGQUIT)
time.sleep(1)
child_process.kill()
print >> sys.stderr, "DataX Process was killed ! you did ?"
sys.exit(RET_STATE["KILL"])
def register_signal():
if not isWindows():
global child_process
signal.signal(2, suicide)
signal.signal(3, suicide)
signal.signal(15, suicide)
def getOptionParser():
usage = "usage: %prog [options] job-url-or-path"
parser = OptionParser(usage=usage)
prodEnvOptionGroup = OptionGroup(parser, "Product Env Options",
"Normal user use these options to set jvm parameters, job runtime mode etc. "
"Make sure these options can be used in Product Env.")
prodEnvOptionGroup.add_option("-j", "--jvm", metavar="<jvm parameters>", dest="jvmParameters", action="store",
default=DEFAULT_JVM, help="Set jvm parameters if necessary.")
prodEnvOptionGroup.add_option("--jobid", metavar="<job unique id>", dest="jobid", action="store", default="-1",
help="Set job unique id when running by Distribute/Local Mode.")
prodEnvOptionGroup.add_option("-m", "--mode", metavar="<job runtime mode>",
action="store", default="standalone",
help="Set job runtime mode such as: standalone, local, distribute. "
"Default mode is standalone.")
prodEnvOptionGroup.add_option("-p", "--params", metavar="<parameter used in job config>",
action="store", dest="params",
help='Set job parameter, eg: the source tableName you want to set it by command, '
'then you can use like this: -p"-DtableName=your-table-name", '
'if you have mutiple parameters: -p"-DtableName=your-table-name -DcolumnName=your-column-name".'
'Note: you should config in you job tableName with ${tableName}.')
prodEnvOptionGroup.add_option("-r", "--reader", metavar="<parameter used in view job config[reader] template>",
action="store", dest="reader",type="string",
help='View job config[reader] template, eg: mysqlreader,streamreader')
prodEnvOptionGroup.add_option("-w", "--writer", metavar="<parameter used in view job config[writer] template>",
action="store", dest="writer",type="string",
help='View job config[writer] template, eg: mysqlwriter,streamwriter')
parser.add_option_group(prodEnvOptionGroup)
devEnvOptionGroup = OptionGroup(parser, "Develop/Debug Options",
"Developer use these options to trace more details of DataX.")
devEnvOptionGroup.add_option("-d", "--debug", dest="remoteDebug", action="store_true",
help="Set to remote debug mode.")
devEnvOptionGroup.add_option("--loglevel", metavar="<log level>", dest="loglevel", action="store",
default="info", help="Set log level such as: debug, info, all etc.")
parser.add_option_group(devEnvOptionGroup)
return parser
def print_json(data):
print(json.dumps(data, sort_keys=True, indent=4, separators=(', ', ': '), ensure_ascii=False))
# nums = {"name": "Mike", "age": 12}
def setCache(data,file_name):
with open(file_name, 'w+') as r:
# '''写入json文件'''
json.dump(data, r)
return data
def getCache(file_name):
with open(file_name,'r') as r:
data = json.load(r) # 返回列表数据，也支持字典
# print("读取json文件：", numbers)
return data
# file 文件名 filenames文件名去所有斜杠用于缓存上次请求时间作为jsonkey，difminutes差值分，每次请求低于*分不会重复执行
def generateJobConfigTemplate(file,filenames,difminutes):
nowtime =NOWTIME.strftime("%Y-%m-%d %H:%M:%S")
readerTemplatePath = DATAX_HOME+'/bin/'+file
readerPar = getCache(readerTemplatePath)
if("difMinutes" in readerPar["job"]):
difminutes= readerPar["job"]["difMinutes"]
minDifMinutes=difminutes/2
if("minDifMinutes" in readerPar["job"]):
minDifMinutes= readerPar["job"]["minDifMinutes"]
# print(NOWTIME)
lastTime=(NOWTIME-datetime.timedelta(minutes=difminutes)).strftime("%Y-%m-%d %H:%M:%S")
try:
cache= getCache(CACHEFile)
except Exception as e:
# 默认1小时前
cache={filenames:lastTime}
setCache(cache,CACHEFile)
if(filenames in cache):
lastTime=cache[filenames]
else:
cache[filenames] =lastTime
setCache(cache, CACHEFile)
startTime = datetime.datetime.strptime(lastTime, "%Y-%m-%d %H:%M:%S")
endTime = (startTime + datetime.timedelta(minutes=difminutes)).strftime("%Y-%m-%d %H:%M:%S")
difs=(datetime.datetime.strptime(nowtime, "%Y-%m-%d %H:%M:%S") - datetime.datetime.strptime(
endTime, "%Y-%m-%d %H:%M:%S")).total_seconds()/60
print(difs,minDifMinutes)
# 最大时间大于当前时间 重新赋值当前时间
if (difs <0):
endTime=nowtime
print(endTime,startTime.strftime("%Y-%m-%d %H:%M:%S"))
difs=(datetime.datetime.strptime(endTime, "%Y-%m-%d %H:%M:%S") - datetime.datetime.strptime(lastTime, "%Y-%m-%d %H:%M:%S")).total_seconds()
# 低于最低差值分钟 不执行
if(difs<(minDifMinutes*60)):
return None
print(difs)
# difs = (datetime.datetime.strptime(endTime, "%Y-%m-%d %H:%M:%S") - datetime.datetime.strptime(
# lastTime, "%Y-%m-%d %H:%M:%S")).total_seconds()
print("上次时间：" + str(lastTime) + "____本次时间:" + str(endTime) + "_____当前时间：" + str(
nowtime) + "执行时间差{" + str(difs) + "秒," + str(difs / 60) + "分," + str(difs / 3600) + "小时," + str(
difs / 86400)+"天")
# try:
if ("where" in readerPar["job"]['content'][0]["reader"]["parameter"]):
where=readerPar["job"]['content'][0]["reader"]["parameter"]["where"]
# slot = re.findall(r"between(.*)\)", where)
# print(slot)
readerPar["job"]['content'][0]["reader"]["parameter"]["where"] = where.replace("(?)"," '"+lastTime+"' and '"+endTime+"' ")
print(readerPar["job"]['content'][0]["reader"]["parameter"]["where"])
if("querySql" in readerPar["job"]['content'][0]["reader"]["parameter"]):
readerPar["job"]['content'][0]["reader"]["parameter"]["querySql"] = where.replace("(?)"," '"+lastTime+"' and '"+endTime+"' ")
print(readerPar["job"]['content'][0]["reader"]["parameter"]["querySql"])
file = readerTemplatePath.split('.json', 1)[0] + 'Start.json'
setCache(readerPar,file)
cache[filenames] =endTime
return {"file":file,"cache":cache}
def readPluginTemplate(plugin):
with open(plugin, 'r') as f:
return json.load(f)
def isUrl(path):
if not path:
return False
assert (isinstance(path, str))
m = re.match(r"^http[s]?://\S+\w*", path.lower())
if m:
return True
else:
return False
def buildStartCommand(options, jobResource):
commandMap = {}
tempJVMCommand = DEFAULT_JVM
if options.jvmParameters:
tempJVMCommand = tempJVMCommand + " " + options.jvmParameters
if options.remoteDebug:
tempJVMCommand = tempJVMCommand + " " + REMOTE_DEBUG_CONFIG
print 'local ip: ', getLocalIp()
if options.loglevel:
tempJVMCommand = tempJVMCommand + " " + ("-Dloglevel=%s" % (options.loglevel))
if options.mode:
commandMap["mode"] = options.mode
# jobResource 可能是 URL，也可能是本地文件路径（相对,绝对）
# jobResource = args[0]
if not isUrl(jobResource):
jobResource = os.path.abspath(jobResource)
if jobResource.lower().startswith("file://"):
jobResource = jobResource[len("file://"):]
jobParams = ("-Dlog.file.name=%s") % (jobResource[-20:].replace('/', '_').replace('.', '_'))
if options.params:
jobParams = jobParams + " " + options.params
if options.jobid:
commandMap["jobid"] = options.jobid
commandMap["jvm"] = tempJVMCommand
commandMap["params"] = jobParams
commandMap["job"] = jobResource
return Template(ENGINE_COMMAND).substitute(**commandMap)
def printCopyright():
print ('''
DataX (%s), From Alibaba !
Copyright (C) 2010-2017, Alibaba Group. All Rights Reserved.
''' % DATAX_VERSION)
sys.stdout.flush()
def testab(e):
print (e)
if __name__ == "__main__":
printCopyright()
# testab(DATAX_VERSION)
parser = getOptionParser()
# args=sys.argv[1:]
options, args = parser.parse_args(sys.argv[1:])
print(options)
# #
if options.reader is not None and options.writer is not None:
# generateJobConfigTemplate(options.reader,options.writer)
print("RET_STATE-----------------------------啊啊啊啊啊啊啊啊啊啊啊-------------------------------")
print(RET_STATE)
sys.exit(RET_STATE['OK'])
if len(args) != 1:
parser.print_help()
sys.exit(RET_STATE['FAIL'])
file=args[0]
filenames=args[0].replace("/","").replace("\\","").replace(".","")
# 两次请求区间时差
difminutes = 5
if len(args)>1:
difminutes=args[1]
else:
difminutes=5
rt= generateJobConfigTemplate(file,filenames,difminutes)
# print(('cache' in rt))
if(rt==None or ('cache' in rt)!=True):
print("-----------------------------间隔时间太少不需要执行-------------------------------")
sys.exit(0)
startCommand = buildStartCommand(options, rt["file"])
print("RET_STATE-----------------------------啊啊啊啊啊啊啊啊啊啊啊------1-------------------------")
print (startCommand)
child_process = subprocess.Popen(startCommand, shell=True)
register_signal()
(stdout, stderr) = child_process.communicate()
print("RET_STATE-----------------------------啊啊啊啊啊啊啊啊啊啊啊---------------2----------------")
print (child_process.returncode)
if(child_process.returncode==0):
# 执行成功缓存本次执行时间
print(rt)
setCache(rt["cache"],CACHEFile)
sys.exit(child_process.returncode)
