如何在Apache NiFi中执行连接SAP HANA的Python脚本并将查询结果输出为CSV文件

别担心，我来一步步帮你解决这个问题！你现在遇到两个核心问题：一是Python脚本的输出逻辑有错误，二是NiFi的Execute Stream Command处理器配置没到位。我们逐个拆解解决：

一、先修正你的Python脚本 #

你的脚本目前有两个明显问题：打开了文件但没实际使用，而且直接把结果列表传给writerow会导致格式错误。另外，我们需要让脚本把结果输出到标准输出（stdout），这样NiFi才能捕获到输出内容。

修正后的完整脚本如下：

import csv
import sys
import pyhdb
import codecs

# 注册CESU-8编码（保留你的原有逻辑）
codecs.register(lambda s: (
    pyhdb.cesu8.CESU8_CODEC_INFO
    if s in {'cesu-8', 'cesu_8'}
    else None
))

try:
    # 连接SAP HANA
    connection = pyhdb.connect(
        host="192.168.xx.xx",
        port=30215,
        user="user",  # 注意你原用户名里的特殊字符，确认是不是拼写错误
        password="pass"
    )
    
    cursor = connection.cursor()
    
    # 格式化查询语句，更易读
    query = """
    SELECT DISTINCT 
        cepc.PRCTR AS suc_sucursal_id,
        cepct.KTEXT AS suc_sucursal_desc,
        csks.BUKRS AS soc_sociedad_id 
    FROM "SAPABAP1"."CEPC" cepc 
    LEFT JOIN "SAPABAP1"."CEPCT" cepct ON cepc.PRCTR = cepct.PRCTR 
    LEFT JOIN "SAPABAP1"."CSKS" csks ON csks.KOSTL = cepc.PRCTR 
    WHERE csks.BUKRS IN ('1000','1002','2000','3000','3001','4000','5000','5001','7000','8000') 
      AND cepct.KTEXT NOT LIKE '%TEST%'
    """
    cursor.execute(query)
    
    # 获取表头和查询结果
    headers = [desc[0] for desc in cursor.description]  # 提取列名作为CSV表头
    rows = cursor.fetchall()
    
    # 将结果输出到标准输出，让NiFi捕获
    writer = csv.writer(sys.stdout)
    writer.writerow(headers)  # 先写入表头
    writer.writerows(rows)    # 批量写入数据行

except Exception as e:
    # 把错误信息输出到标准错误，方便NiFi日志排查
    print(f"执行出错: {str(e)}", file=sys.stderr)
finally:
    # 确保连接关闭
    if 'connection' in locals() and connection.isconnected():
        connection.close()

二、配置NiFi的Execute Stream Command处理器 #

这是你当前卡住的关键环节，按以下步骤配置：

1. Command：填写Python解释器的绝对路径，比如Linux下的/usr/bin/python3，Windows下的C:\Python39\python.exe，不要只写python，避免系统路径找不到解释器。
2. Command Arguments：填写你修正后Python脚本的绝对路径，比如/opt/nifi/scripts/sap_hana_query.py或D:\nifi\scripts\sap_hana_query.py。
3. Capture Output：一定要勾选！这会让NiFi捕获脚本的标准输出，作为后续FlowFile的内容。
4. Capture Error Output：建议勾选，脚本的错误信息会被放到FlowFile的属性里，方便排查问题。
5. Working Directory：可选，填写脚本所在目录，避免相对路径出错。

三、完善你的NiFi数据流 #

推荐的完整数据流链路：

• GenerateFlowFile：用来触发脚本执行（如果需要定期执行，给这个处理器加定时器；一次性执行的话手动触发即可）
• Execute Stream Command：执行修正后的Python脚本
• PutFile：将捕获的输出写入CSV文件，配置好输出目录（比如/opt/nifi/output），文件名可以设为${filename}.csv或者固定名称sucursales3.csv

四、常见问题排查 #

如果还是执行失败，检查这几点：

• 权限问题：NiFi运行的用户有没有权限执行Python脚本、访问SAP HANA的30215端口、读写脚本目录和输出目录？
• 依赖库问题：NiFi服务器上的Python环境有没有安装pyhdb？可以在服务器上执行python3 -c "import pyhdb"验证，没装的话用pip3 install pyhdb安装。
• 日志排查：查看NiFi的nifi-app.log日志文件，里面会有Execute Stream Command的执行细节，包括具体错误信息。

备注：内容来源于stack exchange，提问作者user3319338