SAP BODS中脚本转换CSV至Parquet失败的问题咨询

1. 检查Parquet依赖库是否安装 #

pandas的to_parquet方法必须依赖pyarrow或fastparquet库，默认不会随pandas自带。在BODS使用的Python环境中执行以下命令验证：

/usr/sap/DBO/dataservices/DataQuality/python/bin/python3.7 -c "import pyarrow; print('pyarrow已安装')"

若报错，执行安装命令：

/usr/sap/DBO/dataservices/DataQuality/python/bin/pip3.7 install pyarrow

2. 修正BODS的exec命令参数 #

当前exec调用格式错误，BODS的exec函数需指定Python解释器完整路径，再传入脚本路径作为参数。替换原有exec命令为：

exec('/usr/sap/DBO/dataservices/DataQuality/python/bin/python3.7', '/usr/sap/DBO/dataservices/DataQuality/python/lib/python3.7/site-packages/XXXX.py', '8');

• 第一个参数：Python解释器的完整执行路径（确保BODS运行用户有权限执行）
• 第二个参数：Python脚本的完整路径
• 第三个参数：超时时间（保持8秒即可）

3. 排查路径与权限问题 #

• 确认CSV文件路径/ds_ext_share/BODS_DEV/Output/xxxx.csv真实存在，且BODS服务运行用户（如dbo用户）拥有读权限
• 确认输出路径/ds_ext_share/BODS_DEV/Output/对BODS用户开放写权限，避免因权限不足无法生成Parquet文件
• 检查Python脚本XXXX.py的路径是否正确，且BODS用户拥有读权限

4. 优化Python脚本并添加错误日志 #

修改脚本，增加错误捕获与日志输出，快速定位问题根源：

import sys
import pandas as pd
import traceback

# 可选：若site-packages不在Python默认路径，添加路径
sys.path.append('/usr/sap/DBO/dataservices/DataQuality/python/lib/python3.7/site-packages')

log_path = '/ds_ext_share/BODS_DEV/Output/convert_log.txt'

try:
    # 根据CSV实际编码调整encoding参数（如gbk、utf-8-sig）
    df = pd.read_csv('/ds_ext_share/BODS_DEV/Output/xxxx.csv', encoding='utf-8')
    # 指定Parquet引擎，避免依赖缺失报错
    df.to_parquet('/ds_ext_share/BODS_DEV/Output/xxxx.parquet', engine='pyarrow')
    
    with open(log_path, 'w') as f:
        f.write('CSV转Parquet成功\n')
        f.write(f'处理数据行数：{len(df)}\n')
except Exception as e:
    with open(log_path, 'w') as f:
        f.write('转换失败，错误信息：\n')
        f.write(str(e) + '\n')
        f.write('详细错误堆栈：\n')
        f.write(traceback.format_exc())

5. 补充S3上传逻辑（可选） #

若需直接将Parquet文件上传至S3，先安装boto3库：

/usr/sap/DBO/dataservices/DataQuality/python/bin/pip3.7 install boto3

在脚本中添加上传代码：

import boto3

# 初始化S3客户端（需提前配置AWS凭证：环境变量、IAM角色或~/.aws/credentials）
s3_client = boto3.client('s3')
try:
    s3_client.upload_file(
        '/ds_ext_share/BODS_DEV/Output/xxxx.parquet',
        '你的S3存储桶名称',
        '存储桶内目标路径/xxxx.parquet'
    )
    with open(log_path, 'a') as f:
        f.write('文件已成功上传至S3\n')
except Exception as s3_err:
    with open(log_path, 'a') as f:
        f.write('S3上传失败：\n')
        f.write(str(s3_err) + '\n')

内容的提问来源于stack exchange，提问作者Madhu Reddy