一、批量生成 datax 的 json 文件

编写python代码

1.能够通过python读取mysql的数据库的表

import pymysql
def getDBData(dbName,tableName):
    db_connection = pymysql.connect(
        host="hadoop11",
        port=3306,
        user="root",
        password="123456",
        database='information_schema'
    )
    cursor = db_connection.cursor()
    sql = f"select column_name ,data_type from information_schema.`COLUMNS` where TABLE_SCHEMA = '{dbName}' and table_name = '{tableName}' order by ordinal_position"
    cursor.execute(sql)
    result = cursor.fetchall()
    cursor.close()
    db_connection.close()
    return result

if __name__ == '__main__':
    result = getDBData("spark_project","oms_order")
    print(result)

2.拼接json文件

# -*- coding: utf-8 -*-
import json
import sys
import os
import pymysql

def getDBData(dbName, tableName):
    # 查询mysql的元数据，根据数据库的名字和表的名字查询改表对应的字段和类型
    # 连接数据库（指定charset避免中文乱码）
    conn = pymysql.connect(
        host='hadoop11',
        user='root',
        password='123456',
        database='information_schema',
        charset='utf8mb4',
        cursorclass=pymysql.cursors.DictCursor  # 返回字典格式，更易读
    )
    cursor = conn.cursor()
    sql = """
                SELECT column_name, data_type 
                FROM information_schema.`COLUMNS` 
                WHERE TABLE_SCHEMA = %s AND table_name = %s
                ORDER BY ordinal_position  # 按字段顺序排序
            """
    # 执行sql语句
    cursor.execute(sql, (dbName, tableName))
    result = cursor.fetchall()
    
    cursor.close()
    conn.close()
    
    return result

def getAllCloumnsName(result):
    cloumnName = map(lambda x:x['COLUMN_NAME'],result)
    list1 = list(cloumnName)
    return ",".join(list1)

def getAllClumnsNameAndType(result):
    # 获取列的名字和类型
    mappings = {
        'bigint': 'bigint',
        'varchar': 'string',
        'int': 'int',
        'datetime': 'string',
        'text': 'string',
        'decimal': 'string',
        'date': 'string',
        'timestamp': 'string',
        'varbinary': 'Bytes',
        'double': 'double',
        'time': 'Date'
    }

    list2 = []
    for x in result:
        item = {
            "name": x['COLUMN_NAME'],
            "type": mappings.get(x['DATA_TYPE'], 'string')
        }
        list2.append(item)
    return list2

if __name__ == '__main__':

    # 校验外部参数
    if len(sys.argv) != 3:
        print("请传入数据库的名字和表的名字")
        sys.exit(1)

    dbName = sys.argv[1]
    tableName = sys.argv[2]
    result = getDBData(dbName, tableName)
    print(result)
    cloumn = getAllCloumnsName(result)
    print(cloumn)

    columnAndType = getAllClumnsNameAndType(result)

    # 拼接json 数据 —— 全部改为 Python2 兼容格式
    query_sql = "select %s from %s" % (cloumn, tableName)
    jdbc_url = "jdbc:mysql://hadoop11:3306/%s" % dbName
    hdfs_path = "/user/hive/warehouse/finance.db/ods_jrxd_%s" % tableName
    file_name = tableName
    output_file = "./datax_json/%s.json" % tableName

    # 创建目录（Python2 兼容）
    if not os.path.exists("./datax_json"):
        os.makedirs("./datax_json")

    # 拼接 JSON 数据结构
    jsonData = {
        "job": {
            "setting": {
                "speed": {
                    "channel": 3
                },
                "errorLimit": {
                    "record": 0,
                    "percentage": 0.02
                }
            },
            "content": [
                {
                    "reader": {
                        "name": "mysqlreader",
                        "parameter": {
                            "username": "root",
                            "password": "123456",
                            "connection": [
                                {
                                    "querySql": [
                                        query_sql
                                    ],
                                    "jdbcUrl": [
                                        jdbc_url
                                    ]
                                }
                            ]
                        }
                    },
                    "writer": {
                        "name": "hdfswriter",
                        "parameter": {
                            "defaultFS": "hdfs://hadoop11:8020",
                            "fileType": "text",
                            "path": hdfs_path,
                            "fileName": file_name,
                            "writeMode": "append",
                            "column": columnAndType,
                            "fieldDelimiter": ","
                        }
                    }
                }
            ]
        }
    }

    # 写入文件
    with open(output_file, "w") as f:
        json.dump(jsonData, f)
        
    print("JSON 文件生成成功：%s" % output_file)

运行上述python脚本

python AutoCreateJson.py jrxd channel_info

二、编写一键生成所有表的json的脚本

1.编写生成所有表的json脚本

#/bin/bash
while read x1
do
    python AutoCreateJson.py jrxd $x1
done  < /root/tables.txt

2.给予执行权

chmod 777 create_datax_json.sh

3.执行脚本

./create_datax_json.sh

三、datax 导入数据到hive

1.文件夹/home/scripts/datax_json 下有很多的 json 文件，编写一个脚本，获取该文件夹下的所有文件，并循环执行 datax.py 文件名 

#!/bin/bash

# 定义 JSON 文件所在目录
JSON_DIR="/home/datax_json"

# 定义 datax.py 的路径（请根据你实际路径修改！）
DATAX_PY="/opt/installs/datax/bin/datax.py"

echo "========================================"
echo "开始执行目录下所有 DataX 任务：$JSON_DIR"
echo "========================================"

# 循环遍历所有 .json 文件
for json_file in "$JSON_DIR"/*.json; do
    # 如果目录里没有 json 文件，跳过
    [ -e "$json_file" ] || continue
    
    echo ""
    echo "====> 正在执行：$json_file"
    
    # 执行 DataX 命令
    python "$DATAX_PY" "$json_file"
    
    
done

echo "========================================"
echo "所有任务执行完成！"
echo "========================================"

2.给予执行权

chmod 777 run_all_datax.sh

3.执行脚本

./run_all_datax.sh

四、总结

  这套方案的核心价值：

        告别手动写 SQL，表结构迁移效率提升 90% 以上

        避免人工失误，保证 MySQL 与 Hive 表结构完全一致

        支持批量生成、自定义分区、存储格式等数仓规范

        轻量无依赖，可快速集成到数据平台、调度工具中