PySpark大数据写入DBF文件性能优化指南

import dbf
from datetime import datetime
import os

# 假设collections已通过spark.sql().collect()获取，
# 且每个元素是Spark Row对象或类似字典的结构
# 例如：collections = spark.sql("SELECT JENISKEGIA, JUMLAHUM_A, ... FROM silastik.sakernas_2022_8").collect()

# 模拟数据，实际应为Spark DataFrame.collect()的结果
collections = [
    {'JENISKEGIA': 1, 'JUMLAHUM_A': 100, 'URUTAN': 1, 'WEIGHT': 1.5},
    {'JENISKEGIA': 2, 'JUMLAHUM_A': 200, 'URUTAN': 2, 'WEIGHT': 2.5},
    # ... 更多数据
]
# 为了dbf.write兼容，Spark Row对象需要转换为字典
collections_as_dict = [row.asDict() if hasattr(row, 'asDict') else row for row in collections]


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,2)" # 示例header，请根据实际数据类型和精度调整

new_table = dbf.Table(filename, header)
new_table.open(dbf.READ_WRITE)

for row_data in collections_as_dict:
    new_table.append(row_data) # 逐行追加，每次都可能触发文件元数据更新

new_table.close()
print(f"逐行追加方式写入完成: {filename}")

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

# collections_as_dict 同上
collections_as_dict = [
    {'JENISKEGIA': 1, 'JUMLAHUM_A': 100, 'URUTAN': 1, 'WEIGHT': 1.5},
    {'JENISKEGIA': 2, 'JUMLAHUM_A': 200, 'URUTAN': 2, 'WEIGHT': 2.5},
    # ... 更多数据
]

filename = f"/home/sak202208_{datetime.now().strftime('%Y%m%d%H%M%S')}_multithread_tes.dbf"
header = "JENISKEGIA N(8,0); JUMLAHUM_A N(8,0); URUTAN N(7,0); WEIGHT N(8,2)"

new_table = dbf.Table(filename, header)
new_table.open(dbf.READ_WRITE)

def append_row_task(table, record):
    # 注意：此处的append可能仍然是线程安全的瓶颈
    table.append(record)

# max_workers根据CPU核心数调整
max_workers = min(32, (os.cpu_count() or 1) + 4)
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
    # 提交任务，但实际执行可能因dbf库内部机制而受限
    futures = [executor.submit(append_row_task, new_table, row_data) for row_data in collections_as_dict]
    # 等待所有任务完成
    concurrent.futures.wait(futures)

new_table.close()
print(f"多线程方式写入完成: {filename}")

import dbf
from datetime import datetime
import os

# 假设collections已通过spark.sql().collect()获取，
# 且每个元素是Spark Row对象或类似字典的结构
# 为了dbf.write兼容，Spark Row对象需要转换为字典
collections = [
    {'JENISKEGIA': 1, 'JUMLAHUM_A': 100, 'URUTAN': 1, 'WEIGHT': 1.5},
    {'JENISKEGIA': 2, 'JUMLAHUM_A': 200, 'URUTAN': 2, 'WEIGHT': 2.5},
    {'JENISKEGIA': 3, 'JUMLAHUM_A': 300, 'URUTAN': 3, 'WEIGHT': 3.5},
    {'JENISKEGIA': 4, 'JUMLAHUM_A': 400, 'URUTAN': 4, 'WEIGHT': 4.5},
    {'JENISKEGIA': 5, 'JUMLAHUM_A': 500, 'URUTAN': 5, 'WEIGHT': 5.5},
    # ... 更多数据
]
collections_as_dict = [row.asDict() if hasattr(row, 'asDict') else row for row in collections]


filename = f"/home/sak202208_{datetime.now().strftime('%Y%m%d%H%M%S')}_optimized_tes.dbf"
header = "JENISKEGIA N(8,0); JUMLAHUM_A N(8,0); URUTAN N(7,0); WEIGHT N(8,2)" # 示例header，请根据实际数据类型和精度调整

new_table = dbf.Table(filename, header)
new_table.open(dbf.READ_WRITE)

# 1. 批量预分配空行
number_of_rows = len(collections_as_dict)
if number_of_rows > 0:
    new_table.append(multiple=number_of_rows)

# 2. 迭代记录并原地更新
# 注意：collections_as_dict中的每个元素必须是字典或能作为关键字参数传入dbf.write的映射
for rec, row_data in zip(new_table, collections_as_dict):
    dbf.write(rec, **row_data) # 使用**row_data将字典解包为关键字参数

new_table.close()
print(f"优化方式写入完成: {filename}")