可以通过设置AWS_ACCESS_KEY_ID和AWS_SECRET_ACCESS_KEY的环境变量来解决此问题。以下是示例代码：

import os from airflow.contrib.hooks.emr_hook import EMRHook from airflow.exceptions import AirflowException

设置环境变量

os.environ['AWS_ACCESS_KEY_ID'] = '' os.environ['AWS_SECRET_ACCESS_KEY'] = ''

创建EMRHook实例

emr_hook = EMRHook(emr_conn_id='')

添加步骤

step = emr_hook.add_step( job_flow_id='', step={ 'Name': 'MyStep', 'ActionOnFailure': 'CONTINUE', 'HadoopJarStep': { 'Jar': 's3:///', 'Args': [] } } )

检查步骤是否成功添加

if not step: raise AirflowException("Failed to add step to EMR cluster")