如何在Oozie成功邮件中展示PySpark任务输出表的行数？

原方案失效原因 #

你在PySpark代码中设置的ROW_COUNT环境变量仅存在于PySpark任务运行的YARN容器进程内，而Oozie的邮件动作属于工作流的独立步骤，运行在完全不同的上下文环境中，无法直接读取容器内的环境变量，这是导致需求无法实现的核心原因。

可行方案 #

方案1：通过HDFS文件传递行数

1. 修改PySpark代码：将行数写入Oozie可访问的HDFS路径（或本地共享路径）

row_count = output_table.count()
print(f"Row count of table: {row_count}")
# 写入HDFS文件，建议带上日期避免冲突
with open("/tmp/row_count_${nominal_date}.txt", "w") as f:
    f.write(str(row_count))
# 若使用HDFS API，可替换为：
# from hdfs import InsecureClient
# client = InsecureClient('http://namenode:50070', user='oozie')
# client.write(f'/user/oozie/row_count_{nominal_date}.txt', str(row_count))

2. 修改workflow.xml：添加fs动作读取文件内容为Oozie变量，再在邮件中引用

<!-- 新增读取行数的动作 -->
<action name="readRowCount">
    <fs>
        <cat path="/tmp/row_count_${nominal_date}.txt" to="ROW_COUNT"/>
    </fs>
    <ok to="sendSuccessEmail"/>
    <error to="kill"/>
</action>

<!-- 修改邮件动作中的行数引用 -->
<action name="sendSuccessEmail">
    <email xmlns="uri:oozie:email-action:0.1">
        <to>${successEmailTo}</to>
        <subject>${clusterName}: Workflow succeeded for the date: ${nominal_date}</subject>
        <body>
Hi Team,

The workflow completed successfully for the date ${nominal_date}.
    Workflow details
    ----------------
    Cluster Name:   ${clusterName}
    Nominal time:   ${nominal_date}
    End time:       ${timestamp()}
    Row count:      ${ROW_COUNT}
    
Note: This is a auto generated email, for any further details please contact our DE team.
        </body>
    </email>
    <ok to="end"/>
    <error to="kill"/>
</action>

注意：需确保Oozie有该文件路径的读写权限，同时可在kill节点添加清理文件的逻辑，避免冗余文件积累。

方案2：通过Spark动作捕获输出变量

1. 修改PySpark代码：按固定格式打印行数到标准输出

row_count = output_table.count()
# 必须以"变量名=值"的格式打印，Oozie会自动解析
print(f"ROW_COUNT={row_count}")

2. 修改workflow.xml：在Spark动作中开启capture-output

<action name="runPySpark">
    <spark xmlns="uri:oozie:spark-action:0.2">
        <job-tracker>${jobTracker}</job-tracker>
        <name-node>${nameNode}</name-node>
        <master>yarn</master>
        <mode>cluster</mode>
        <spark-opts>--deploy-mode cluster</spark-opts>
        <app-jar>hdfs:///path/to/your/pyspark_script.py</app-jar>
        <capture-output/> <!-- 开启输出捕获，自动解析标准输出中的变量 -->
    </spark>
    <ok to="sendSuccessEmail"/>
    <error to="kill"/>
</action>

此时邮件动作中直接使用${ROW_COUNT}即可引用行数，无需额外步骤。

方案3：通过数据库存储与查询传递

1. 修改PySpark代码：将行数写入Hive/MySQL等状态表

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("RowCountTracker").getOrCreate()
row_count = output_table.count()
# 插入或更新状态表
spark.sql(f"""
    INSERT INTO workflow_status (nominal_date, row_count)
    VALUES ('${nominal_date}', {row_count})
    ON DUPLICATE KEY UPDATE row_count={row_count}
""")

2. 修改workflow.xml：添加Hive动作查询行数并赋值

<action name="fetchRowCount">
    <hive xmlns="uri:oozie:hive-action:0.2">
        <job-tracker>${jobTracker}</job-tracker>
        <name-node>${nameNode}</name-node>
        <script>hdfs:///path/to/fetch_row_count.hql</script>
        <param>NOMINAL_DATE=${nominal_date}</param>
        <capture-output/>
    </hive>
    <ok to="sendSuccessEmail"/>
    <error to="kill"/>
</action>

对应的fetch_row_count.hql内容：

SELECT row_count FROM workflow_status WHERE nominal_date = '${NOMINAL_DATE}'

邮件中直接使用${row_count}引用即可。

额外提示 #

原邮件配置中的suceeded存在拼写错误，建议修正为succeeded。

内容的提问来源于stack exchange，提问作者Anusha Ravindra