如何使用Airflow在DataProc集群上执行HDFS复制命令

当你已经通过Airflow完成Dataproc集群创建流程，需要把存储在Google Cloud Storage上的JAR文件同步到集群HDFS的指定目录时，不需要额外搭建SSH通道、手动安装依赖，直接基于Airflow内置的Dataproc生态操作符即可完成全流程自动化，具体实现如下：

实现前提 #

• 已通过Airflow的Dataproc创建操作符拿到目标集群的标识信息：项目ID、部署区域、集群名称
• Airflow绑定的运行服务账号已配置两类权限：可向目标Dataproc集群提交Hadoop作业的权限、可读取源GCS存储桶对象的权限

具体DAG编写逻辑 #

• 用Airflow内置的DataprocHadoopOperator作为拷贝任务的执行主体，这个操作符会直接向Dataproc集群提交原生DistCp作业，依托Dataproc自带的GCS连接器完成gs://协议解析，不需要在节点上额外配置GCS访问凭证
• 给操作符配置DistCp的执行参数：
  • 源路径：按照实际GCS地址填写，格式为gs://你的存储桶名/待同步JAR所在路径/*.jar，用通配符精确匹配需要拷贝的JAR文件，避免同步无关内容
  • 目标路径：对应HDFS上的目标目录，格式为hdfs:///你的HDFS目标目录路径，比如常规存放作业依赖JAR的路径hdfs:///user/hadoop/spark-jars/
  • 作业优化参数：追加-update参数自动跳过HDFS上已存在的同名JAR、-m 10参数限制10个并发拷贝任务，避免小文件过多占用集群计算资源
• 注意：DistCp写入的HDFS路径是集群全局共享的存储路径，不需要额外执行“先拷贝到主节点本地磁盘、再上传到HDFS”的冗余流程，直接配置HDFS目标路径即可完成文件同步，大幅减少拷贝耗时
• 在DAG的依赖配置中必须明确绑定任务执行顺序：Dataproc集群创建任务 → JAR拷贝任务，从流程上避免作业提交时集群未完成初始化导致的连接超时、资源未就绪类报错

常见问题排查 #

• 不要选择SSH连接主节点执行拷贝命令的方案：手动SSH需要在Airflow节点配置集群主节点的网络连通性、密钥权限，后续集群IP变更、网络策略调整时很容易断连，官方内置的作业提交通道稳定性远高于自建SSH方案
• 提前校验HDFS目标目录的写权限：Dataproc默认hadoop用户对/user/hadoop路径下的目录有完整写权限，如果需要写入其他HDFS路径，记得在集群初始化阶段通过初始化动作提前创建目录并配置对应写权限，否则会出现DistCp作业权限拒绝报错
• 避免通配符匹配范围过大：如果源GCS目录下同时存在其他类型的文件，一定要在通配符路径中加上.jar后缀限定同步范围，防止无关文件被冗余拷贝到HDFS占用存储

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