通过HDFS HTTP Web API执行PUT请求触发BrokenPipeError问题求助

1. Location地址不变的问题 #

你当前使用的是HttpFS代理服务而非原生WebHDFS服务，HttpFS作为HDFS的HTTP代理层，会负责转发所有请求到后端DataNode，不会直接返回DataNode的地址给客户端，因此第一次请求返回的307响应Location字段和初始请求地址一致是正常现象，不需要跳转到DataNode。

2. BrokenPipeError报错原因 #

• 低级代码错误：你在第二次发送数据的PUT请求中，重复携带了op=CREATE等参数。第一次请求返回的Location地址已经包含了所有必要的签名、操作参数，重复传参会导致服务端参数解析异常，主动断开连接。
• 服务端配置限制：HttpFS默认的最大请求大小通常为128M，你的200M文件超过默认阈值，服务端会直接中断连接。
• 超时配置不合理：默认的请求超时时间过短，大文件传输过程中触发超时断开。
• 分块追加方式本身有额外开销：每次追加都需要重新走认证、元数据校验流程，不仅速度慢，还会增加连接异常概率。

1. 先调整服务端配置（联系集群管理员操作） #

调整以下HttpFS配置参数，适配200M级别的文件传输：

<!-- 最大请求大小调整为512M，适配你的文件大小 -->
<property>
  <name>httpfs.max.request.size</name>
  <value>536870912</value>
</property>
<!-- Socket超时调整为10分钟 -->
<property>
  <name>httpfs.socket.timeout</name>
  <value>600000</value>
</property>
<!-- 连接超时调整为30秒 -->
<property>
  <name>httpfs.connection.timeout</name>
  <value>30000</value>
</property>

如果集群允许，优先改用原生WebHDFS服务，直接和DataNode通信，性能可以达到和hdfs dfs -put相近的水平。

2. 修正客户端代码 #

修正参数重复问题，增加连接复用、超时、块大小配置，避免分块追加：

import requests
from requests_kerberos import HTTPKerberosAuth, OPTIONAL

class HttpFS:
    def __init__(self, url=settings.HTTPFS_URL):
        self.httpfs = url
        self.auth = HTTPKerberosAuth(mutual_authentication=OPTIONAL)
        # 复用Session减少Kerberos认证开销
        self.session = requests.Session()
        self.session.verify = settings.CA_ROOT_PATH
        self.session.auth = self.auth
        # 超时时间设为5分钟，可根据需要调整
        self.timeout = 300

    def put(self, local_file, hdfs_dir):
        url = f"{self.httpfs}{hdfs_dir}"
        params = {"op": "CREATE", "overwrite": True}
        # 第一次请求拿Location
        resp = self.session.put(
            url, 
            params=params, 
            allow_redirects=False,
            timeout=self.timeout
        )
        resp.raise_for_status()
        redirect_url = resp.headers['Location']
        # 第二次传数据不要带params，redirect_url已经包含所有必要参数
        with open(local_file, 'rb') as f:
            resp = self.session.put(
                redirect_url,
                data=f,
                stream=True,
                timeout=self.timeout
            )
        resp.raise_for_status()

如果还是偶发BrokenPipe，可以调整发送块大小，迭代分块上传：

with open(local_file, 'rb') as f:
    resp = self.session.put(
        redirect_url,
        data=iter(lambda: f.read(1024*1024), b''), # 每次读1M上传
        stream=True,
        timeout=self.timeout
    )

3. 可选优化 #

• 如果Kerberos双向认证导致异常，可以把mutual_authentication参数改为DISABLED，多数HttpFS服务不需要双向认证。
• 避免在传输高峰期发起大文件请求，减少集群带宽不足导致的断连。

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