Python 获取 datax 执行结果保存到数据库的方法

下面是关于Python获取datax执行结果保存到数据库的完整攻略：

1. 前置工作

首先需要安装好datax和对应数据库的驱动包，以及Python所需的相关库。

2. 编写Python代码

2.1 准备datax执行配置文件

先准备好要执行的datax配置文件，例如 job.json 文件。

2.2 执行datax作业并获取执行结果

执行命令：

python /path/to/datax/bin/datax.py /path/to/job.json

在Python代码中，可以通过 os.system() 函数执行对应命令并获取执行结果。

import os

command = 'python /path/to/datax/bin/datax.py /path/to/job.json'

result = os.popen(command).read()

其中，result 是获取到的执行结果字符串。

2.3 解析执行结果

可以使用 re 模块或其他方式解析 result 字符串，获得需要的信息。

例如，如果希望获取datax执行的总记录数，可以使用如下正则表达式：

import re

# 解析总记录数
pattern = r'.*?(\d+) all.'
total_records = re.match(pattern, result).group(1)

2.4 保存解析结果到数据库

最后，将解析结果保存到数据库中。这里以MySQL为例，使用 pymysql 模块实现：

import pymysql

# 连接MySQL数据库
conn = pymysql.connect(host='localhost', port=3306, user='root', password='password', db='test')

# 创建游标
cursor = conn.cursor()

# 执行SQL插入语句
sql = 'INSERT INTO result(total_records) VALUES (%s)'
cursor.execute(sql, (total_records,))

# 提交事务
conn.commit()

# 关闭游标和连接
cursor.close()
conn.close()

这样，就将datax执行结果中的总记录数保存到MySQL数据库中了。

3. 示例说明

示例一

假设我们有一个datax作业，它的配置文件为 job.json，数据源是MySQL，需要将数据导入到Elasticsearch中。我们可以编写如下Python代码：

import os
import re
import pymysql

# 执行datax作业
command = 'python /path/to/datax/bin/datax.py /path/to/job.json'
result = os.popen(command).read()

# 解析执行结果
# 获取总记录数
pattern = r'.*?(\d+) all.'
total_records = re.match(pattern, result).group(1)

# 保存总记录数到MySQL数据库中
conn = pymysql.connect(host='localhost', port=3306, user='root', password='password', db='test')
cursor = conn.cursor()
sql = 'INSERT INTO result(total_records) VALUES (%s)'
cursor.execute(sql, (total_records,))
conn.commit()
cursor.close()
conn.close()

这样，我们就实现了将datax执行结果中的总记录数保存到MySQL数据库中。

示例二

假设我们有一个datax作业，它的配置文件为 job.json，数据源是Oracle，需要将数据导入到Hadoop中。我们可以编写如下Python代码：

import os
import re
import pymysql

# 执行datax作业
command = 'python /path/to/datax/bin/datax.py /path/to/job.json'
result = os.popen(command).read()

# 解析执行结果
# 获取总记录数
pattern = r'TOTAL\sRECORDS:\s+(.*?)$'
total_records_match = re.search(pattern, result, re.MULTILINE)
if total_records_match:
    total_records = total_records_match.group(1)
else:
    total_records = 0

# 获取导入成功记录数
pattern = r'TOTAL\sBYTES:\s+(.*?)$'
total_bytes_match = re.search(pattern, result, re.MULTILINE)
if total_bytes_match:
    total_bytes = total_bytes_match.group(1)
else:
    total_bytes = 0

# 保存解析结果到MySQL数据库中
conn = pymysql.connect(host='localhost', port=3306, user='root', password='password', db='test')
cursor = conn.cursor()
sql = 'INSERT INTO result(total_records, total_bytes) VALUES (%s, %s)'
cursor.execute(sql, (total_records, total_bytes))
conn.commit()
cursor.close()
conn.close()

这样，我们就实现了将datax执行结果中的总记录数和导入成功记录数保存到MySQL数据库中。