PySpark高效写入DBF文件的策略与优化

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

# 假设collections已通过spark.sql(...).collect()获取
# collections = spark.sql("SELECT JENISKEGIA, JUMLAHUM_A, ... , URUTAN, WEIGHT FROM silastik.sakernas_2022_8").collect()

# 模拟数据，实际应用中替换为Spark DataFrame的collect结果
collections = [
    {'JENISKEGIA': 1, 'JUMLAHUM_A': 100, 'URUTAN': 1, 'WEIGHT': 10.5},
    {'JENISKEGIA': 2, 'JUMLAHUM_A': 200, 'URUTAN': 2, 'WEIGHT': 20.1},
    # ... 更多数据
]

filename_base = "/home/sak202208_tes.dbf"
filename = filename_base.replace(".dbf", f"_{datetime.now().strftime('%Y%m%d%H%M%S')}.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_data in collections:
    new_table.append(row_data) # 每次append都会触发类型转换和文件I/O

new_table.close()
print(f"传统写入完成: {filename}")

# 尝试多线程写入（通常效果不佳）
# 注意：dbf库的append操作可能不是线程安全的，或因底层文件锁导致竞争
# filename_mt = filename_base.replace(".dbf", f"_mt_{datetime.now().strftime('%Y%m%d%H%M%S')}.dbf")
# new_table_mt = dbf.Table(filename_mt, header)
# new_table_mt.open(dbf.READ_WRITE)

# def append_row(table_obj, record):
#     table_obj.append(record) # 这里的append依然是逐条操作

# with concurrent.futures.ThreadPoolExecutor(max_workers=min(32, (os.cpu_count() or 1) + 4)) as executor:
#     futures = [executor.submit(append_row, new_table_mt, row_data) for row_data in collections]
#     for future in concurrent.futures.as_completed(futures):
#         try:
#             future.result()
#         except Exception as exc:
#             print(f'生成异常: {exc}')

# new_table_mt.close()
# print(f"多线程写入完成: {filename_mt}")

import dbf
from datetime import datetime

# 假设collections已通过spark.sql(...).collect()获取
# collections = spark.sql("SELECT JENISKEGIA, JUMLAHUM_A, ... , URUTAN, WEIGHT FROM silastik.sakernas_2022_8").collect()

# 模拟数据，实际应用中替换为Spark DataFrame的collect结果
# 注意：从Spark Row对象转换为字典或命名元组更利于dbf.write(**row)
collections_for_optimized = [
    {'JENISKEGIA': 1, 'JUMLAHUM_A': 100, 'URUTAN': 1, 'WEIGHT': 10.5},
    {'JENISKEGIA': 2, 'JUMLAHUM_A': 200, 'URUTAN': 2, 'WEIGHT': 20.1},
    {'JENISKEGIA': 3, 'JUMLAHUM_A': 300, 'URUTAN': 3, 'WEIGHT': 30.2},
    {'JENISKEGIA': 4, 'JUMLAHUM_A': 400, 'URUTAN': 4, 'WEIGHT': 40.3},
    {'JENISKEGIA': 5, 'JUMLAHUM_A': 500, 'URUTAN': 5, 'WEIGHT': 50.4},
    # ... 更多数据
]


filename_optimized_base = "/home/sak202208_optimized_tes.dbf"
filename_optimized = filename_optimized_base.replace(".dbf", f"_{datetime.now().strftime('%Y%m%d%H%M%S')}.dbf")

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

new_table_optimized = dbf.Table(filename_optimized, header_optimized)
new_table_optimized.open(dbf.READ_WRITE)

# 1. 预分配所有行
# 需要知道总行数，这里假设collections_for_optimized的长度就是总行数
num_rows = len(collections_for_optimized)
if num_rows > 0:
    new_table_optimized.append(multiple=num_rows)

# 2. 遍历并填充数据
# 注意：collections中的每个row必须是字典（或类似mapping的对象），
# 才能与dbf.write(rec, **row)配合使用。
# Spark的Row对象可以直接转换为字典：row.asDict()
for rec, row_data in zip(new_table_optimized, collections_for_optimized):
    dbf.write(rec, **row_data) # 使用**row_data将字典解包为关键字参数

new_table_optimized.close()
print(f"优化写入完成: {filename_optimized}")