執(zhí)行 datax 作業(yè)，創(chuàng)建執(zhí)行文件，在 crontab 中每天1點（下面有關(guān)系）執(zhí)行：
其中 job_start 及 job_finish 這兩行記錄是自己添加的，為了方便識別出哪張表。
#!/bin/bash
source /etc/profile
user1="root"
pass1="pwd"
user2="root"
pass2="pwd"
job_path="/opt/datax/job/"
jobfile=(
job_table_a.json
job_table_b.json
)
for filename in ${jobfile[@]}
do
echo "job_start: "`date "+%Y-%m-%d %H:%M:%S"`" ${filename}"
python /opt/datax/bin/datax.py -p "-Duser1=${user1} -Dpass1=${pass1} -Duser2=${user2} -Dpass2=${pass2}" ${job_path}${filename}
echo "job_finish: "`date "+%Y-%m-%d %H:%M:%S"`" ${filename}"
done
# 0 1 * * * /opt/datax/job/dc_to_ods_incr.sh >> /opt/datax/job/log/dc_to_ods_incr_$(date +\%Y\%m\%d_\%H\%M\%S).log 2>&1
# egrep '任務(wù)|速度|總數(shù)|job_start|job_finish' /opt/datax/job/log/
datax 執(zhí)行日志：
job_start: 2018-08-08 01:13:28 job_table_a.json
任務(wù)啟動時刻 : 2018-08-08 01:13:28
任務(wù)結(jié)束時刻 : 2018-08-08 01:14:49
任務(wù)總計耗時 : 81s
任務(wù)平均流量 : 192.82KB/s
記錄寫入速度 : 1998rec/s
讀出記錄總數(shù) : 159916
讀寫失敗總數(shù) : 0
job_finish: 2018-08-08 01:14:49 job_table_a.json
job_start: 2018-08-08 01:14:49 job_table_b.json
任務(wù)啟動時刻 : 2018-08-08 01:14:50
任務(wù)結(jié)束時刻 : 2018-08-08 01:15:01
任務(wù)總計耗時 : 11s
任務(wù)平均流量 : 0B/s
記錄寫入速度 : 0rec/s
讀出記錄總數(shù) : 0
讀寫失敗總數(shù) : 0
job_finish: 2018-08-08 01:15:01 job_table_b.json
接下來讀取這些信息保存到數(shù)據(jù)庫，在數(shù)據(jù)庫中創(chuàng)建表：
CREATE TABLE `datax_job_result` (
`log_file` varchar(200) DEFAULT NULL,
`job_file` varchar(200) DEFAULT NULL,
`start_time` datetime DEFAULT NULL,
`end_time` datetime DEFAULT NULL,
`seconds` int(11) DEFAULT NULL,
`traffic` varchar(50) DEFAULT NULL,
`write_speed` varchar(50) DEFAULT NULL,
`read_record` int(11) DEFAULT NULL,
`failed_record` int(11) DEFAULT NULL,
`job_start` varchar(200) DEFAULT NULL,
`job_finish` varchar(200) DEFAULT NULL,
`insert_time` datetime DEFAULT CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
定時執(zhí)行以下文件，因為 datax 作業(yè) 1 點執(zhí)行，為了獲取一天內(nèi)最新生產(chǎn)的日志，腳本中取 82800內(nèi)生產(chǎn)的日志文件，及23 小時內(nèi)生產(chǎn)的那個最新日志。所以一天內(nèi)任何時間執(zhí)行都可以。此文件也是定時每天執(zhí)行（判斷 datax 作業(yè)完成后執(zhí)行）
#!/usr/bin/python
# -*- coding: UTF-8 -*-
# 0 5 * * * source /etc/profile && /usr/bin/python2.7 /opt/datax/job/save_log_to_db.py > /dev/null 2>&1
import re
import os
import sqlalchemy
import pandas as pd
import datetime as dt
def save_to_db(df):
engine = sqlalchemy.create_engine("mysql+pymysql://root:pwd@localhost:3306/test", encoding="utf-8")
df.to_sql("datax_job_result", engine, index=False, if_exists='append')
def get_the_latest_file(path):
t0 = dt.datetime.utcfromtimestamp(0)
d2 = (dt.datetime.now() - t0).total_seconds()
d1 = d2 - 82800
for (dirpath, dirnames, filenames) in os.walk(path):
for filename in sorted(filenames, reverse = True):
if filename.endswith(".log"):
f = os.path.join(dirpath,filename)
ctime = os.stat(f)[-1]
if ctime>=d1 and ctime <=d2:
return f
def get_job_result_from_logfile(path):
result = pd.DataFrame(columns=['log_file','job_file','start_time','end_time','seconds','traffic','write_speed','read_record','failed_record','job_start','job_finish'])
log_file = get_the_latest_file(path)
index = 0
content = open(log_file, "r")
for line in content:
result.loc[index, 'log_file'] = log_file
if re.compile(r'job_start').match(line):
result.loc[index, 'job_file'] = line.split(' ')[4].strip()
result.loc[index, 'job_start'] = line,
elif re.compile(r'任務(wù)啟動時刻').match(line):
result.loc[index, 'start_time'] = line.split('刻')[1].strip().split(' ')[1].strip() + ' ' + line.split('刻')[1].strip().split(' ')[2].strip()
elif re.compile(r'任務(wù)結(jié)束時刻').match(line):
result.loc[index, 'end_time'] = line.split('刻')[1].strip().split(' ')[1].strip() + ' ' + line.split('刻')[1].strip().split(' ')[2].strip()
elif re.compile(r'任務(wù)總計耗時').match(line):
result.loc[index, 'seconds'] = line.split(':')[1].strip().replace('s','')
elif re.compile(r'任務(wù)平均流量').match(line):
result.loc[index, 'traffic'] = line.split(':')[1].strip()
elif re.compile(r'記錄寫入速度').match(line):
result.loc[index, 'write_speed'] = line.split(':')[1].strip()
elif re.compile(r'讀出記錄總數(shù)').match(line):
result.loc[index, 'read_record'] = line.split(':')[1].strip()
elif re.compile(r'讀寫失敗總數(shù)').match(line):
result.loc[index, 'failed_record'] = line.split(':')[1].strip()
elif re.compile(r'job_finish').match(line):
result.loc[index, 'job_finish'] = line,
index = index + 1
else:
pass
save_to_db(result)
get_job_result_from_logfile("/opt/datax/job/log")
以上這篇Python 獲取 datax 執(zhí)行結(jié)果保存到數(shù)據(jù)庫的方法就是小編分享給大家的全部內(nèi)容了，希望能給大家一個參考，也希望大家多多支持腳本之家。
更多文章、技術(shù)交流、商務(wù)合作、聯(lián)系博主
微信掃碼或搜索：z360901061
微信掃一掃加我為好友
QQ號聯(lián)系： 360901061
您的支持是博主寫作最大的動力，如果您喜歡我的文章，感覺我的文章對您有幫助，請用微信掃描下面二維碼支持博主2元、5元、10元、20元等您想捐的金額吧，狠狠點擊下面給點支持吧，站長非常感激您！手機微信長按不能支付解決辦法：請將微信支付二維碼保存到相冊，切換到微信，然后點擊微信右上角掃一掃功能，選擇支付二維碼完成支付。
【本文對您有幫助就好】元
