优化PySpark将Hadoop数据写入DBF文件的性能

import dbf
from datetime import datetime
import os

# 假设 spark 变量已初始化
# 从Hive查询数据，并使用collect()将所有结果拉取到Driver内存
collections = spark.sql("SELECT JENISKEGIA, JUMLAHUM_A, ..., URUTAN, WEIGHT FROM silastik.sakernas_2022_8").collect()

filename = f"/home/sak202208_{datetime.now().strftime('%Y%m%d%H%M%S')}_tes.dbf"

# 定义DBF文件结构
header = "JENISKEGIA N(8,0); JUMLAHUM_A N(8,0); ..., URUTAN N(7,0); WEIGHT N(8,0)"

new_table = dbf.Table(filename, header)

new_table.open(dbf.READ_WRITE)

# 逐行遍历并追加
for row in collections:
    new_table.append(row) # 每次append都会触发文件操作和类型转换

new_table.close()

import dbf
from datetime import datetime
import os
import concurrent.futures

# 假设 spark 变量已初始化
collections = spark.sql("SELECT JENISKEGIA, JUMLAHUM_A, ..., URUTAN, WEIGHT FROM silastik.sakernas_2022_8").collect()

filename = f"/home/sak202208_{datetime.now().strftime('%Y%m%d%H%M%S')}_tes.dbf"

header = "JENISKEGIA N(8,0); JUMLAHUM_A N(8,0); ..., URUTAN N(7,0); WEIGHT N(8,0)"

new_table = dbf.Table(filename, header)

new_table.open(dbf.READ_WRITE)

def append_row(table_obj, record_data):
    # 注意：dbf库的append操作并非完全线程安全，且Python GIL会限制CPU密集型任务的并行度
    table_obj.append(record_data)

# 使用线程池提交任务
with concurrent.futures.ThreadPoolExecutor(max_workers=min(32, (os.cpu_count() or 1) + 4)) as executor:
    for row in collections:
        # executor.submit(append_row, new_table, row) # 实际可能因GIL和文件锁导致性能提升不明显
        # 错误示范：此处的append_row(new_table, row)会在主线程中立即执行，而不是提交给线程池
        executor.submit(append_row, new_table, row)

new_table.close()

import dbf
from datetime import datetime
import os

# 假设 spark 变量已初始化
# 从Hive查询数据，并使用collect()将所有结果拉取到Driver内存
# 注意：Spark的Row对象通常可以通过其字段名像字典一样访问，这符合dbf.write的要求
collections = spark.sql("SELECT JENISKEGIA, JUMLAHUM_A, ..., URUTAN, WEIGHT FROM silastik.sakernas_2022_8").collect()

filename = f"/home/sak202208_{datetime.now().strftime('%Y%m%d%H%M%S')}_optimized.dbf"

header = "JENISKEGIA N(8,0); JUMLAHUM_A N(8,0); ..., URUTAN N(7,0); WEIGHT N(8,0)"

new_table = dbf.Table(filename, header)

new_table.open(dbf.READ_WRITE)

# 1. 批量预分配所有记录空间
# 获取需要写入的行数
number_of_rows = len(collections)
if number_of_rows > 0:
    new_table.append(multiple=number_of_rows) # 一次性创建所有空行

# 2. 遍历预分配的记录并填充数据
# zip函数将dbf.Table对象（可迭代，返回记录对象）与Spark Row集合配对
for rec, row in zip(new_table, collections):
    # dbf.write()方法直接将映射（如字典或Spark Row对象）的数据写入到记录中
    # **row 会将Spark Row对象的字段名和值作为关键字参数传递
    dbf.write(rec, **row.asDict()) # 确保row是一个映射，这里将Spark Row转换为字典
    # 如果Spark Row对象本身支持**解包，可以直接 dbf.write(rec, **row)
    # 但为了兼容性，推荐使用 .asDict()

new_table.close()