使用 Airflow 触发 EMR Step 执行失败问题排查与解决

在使用 Airflow 触发 EMR Step 执行时，可能会遇到 "java.io.IOException: Cannot run program 's3://shell script path': error=2, No such file or directory" 错误。这通常是由于使用了 command-runner.jar 尝试直接执行 S3 上的 shell 脚本引起的。本文将深入探讨此问题的原因，并提供解决方案，同时还会提供一种更高效的替代方案，以避免不必要的 EMR 资源消耗。

问题分析

错误信息表明，EMR 无法找到并执行位于 S3 路径 s3://shell script path 的脚本。尽管文件存在且 EMR 实例配置和 Service Role 拥有 S3 的完全访问权限，但直接使用 command-runner.jar 运行 S3 上的脚本是不正确的。

command-runner.jar 主要用于执行 Hadoop 命令或 Java 程序，而不能直接执行 shell 脚本。因此，需要使用 script-runner.jar 来执行 S3 上的 shell 脚本。

解决方案：使用 script-runner.jar

script-runner.jar 是 EMR 提供的用于执行脚本的工具。它会从指定位置下载脚本，并在 EMR 集群上执行。

以下是在 Airflow 中使用 script-runner.jar 的示例代码：

def add_step(cluster_id, script_path):
    response = client.add_job_flow_steps(
        JobFlowId=cluster_id,
        Steps=[
            {
                'Name': 'Run Script from S3',
                'ActionOnFailure': 'CONTINUE',
                'HadoopJarStep': {
                    'Jar': 's3://us-west-2.elasticmapreduce/libs/script-runner/script-runner.jar', # 替换为你的区域
                    'Args': [script_path]
                }
            },
        ]
    )
    return response['StepIds'][0]

dag = DAG(
    dag_id="EMR_START_DAG",
    description="Trial for EMR start",
    start_date=days_ago(1)
)

EMR_STEP_1 = PythonOperator(
    task_id='EMR_STEP_1',
    python_callable=add_step,
    op_kwargs={'cluster_id': '{{ti.xcom_pull("EMR_START")["JobFlowId"]}}',
               'script_path': 's3://shell script path'},
    dag=dag
)

注意事项：

Natural Language Playlist

探索语言和音乐之间丰富而复杂的关系，并使用 Transformer 语言模型构建播放列表。

67

查看详情

• 确保 script-runner.jar 的路径与你的 EMR 集群所在区域匹配。
• script_path 参数应为 S3 上脚本的完整路径。
• EMR 实例角色需要有权限访问 S3 上的脚本文件。

替代方案：在 Airflow 中直接运行 AWS CLI

如果脚本的功能仅仅是从外部下载数据并上传到 S3，可以考虑直接在 Airflow 中运行 AWS CLI 命令，而无需启动 EMR 集群。这样可以避免不必要的资源消耗。

首先，确保 Airflow 服务器上安装了 AWS CLI，并配置了正确的 AWS 凭证。

然后，可以使用 BashOperator 在 Airflow 中执行 AWS CLI 命令：

from airflow.operators.bash_operator import BashOperator

dag = DAG(
    dag_id="EMR_START_DAG",
    description="Trial for EMR start",
    start_date=days_ago(1)
)

DOWNLOAD_AND_UPLOAD = BashOperator(
    task_id='download_and_upload',
    bash_command='wget -O - https://archive.ics.uci.edu/ml/machine-learning-databases/iris/iris.data | aws s3 cp - s3path',
    dag=dag
)

注意事项：

• 确保 Airflow 服务器具有访问外部网站和 S3 的权限。
• 替换 s3path 为你的 S3 目标路径。
• 根据实际情况调整命令和参数。

总结

当在使用 Airflow 触发 EMR Step 执行时遇到 "java.io.IOException: Cannot run program" 错误时，首先应检查是否使用了正确的 Jar 文件。对于执行 S3 上的 shell 脚本，应使用 script-runner.jar。此外，如果脚本的功能相对简单，可以考虑直接在 Airflow 中运行 AWS CLI 命令，以避免不必要的 EMR 资源消耗。选择哪种方案取决于脚本的复杂度和对 EMR 集群资源的需求。

以上就是使用 Airflow 触发 EMR Step 执行失败问题排查与解决的详细内容，更多请关注php中文网其它相关文章！