mysql到hdfs

import pymysql
from hdfs import InsecureClient

# 连接MySQL
mysql_conn = pymysql.connect(host='localhost', user='user', password='password', db='database')
cursor = mysql_conn.cursor()

# 连接HDFS
hdfs_client = InsecureClient('http://localhost:50070')

# 查询MySQL数据
cursor.execute("SELECT * FROM table")
rows = cursor.fetchall()

# 写入HDFS
with hdfs_client.write('/path/to/file.csv', encoding='utf-8') as writer:
    for row in rows:
        writer.write(','.join(map(str, row)) + '\n')

# 关闭连接
cursor.close()
mysql_conn.close()

# 示例：将MySQL的INT类型转换为HDFS的STRING类型
for row in rows:
    converted_row = [str(item) if isinstance(item, int) else item for item in row]
    writer.write(','.join(converted_row) + '\n')

# 示例：使用多线程并行处理数据迁移
import threading

def migrate_data(start, end):
    # 迁移数据的逻辑
    pass

threads = []
for i in range(0, len(rows), 1000):
    t = threading.Thread(target=migrate_data, args=(i, i+1000))
    threads.append(t)
    t.start()

for t in threads:
    t.join()