如何为通过Java程序启动的Spark Worker配置IP访问过滤？

是的，你可以通过扩展Spark的配置和自定义组件来实现硬编码IP地址列表的访问限制，主要需要针对Web UI服务和RPC通信服务两个核心部分分别处理：

1. Web UI的IP过滤（基于Jetty Servlet过滤器） #

Spark Worker的Web UI基于Jetty运行，你可以通过自定义Servlet Filter拦截请求并校验客户端IP：

步骤1：实现IP过滤Filter #

创建一个自定义Filter类，直接在代码中硬编码允许的IP列表：

import javax.servlet.*;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.HashSet;
import java.util.Set;

public class IPAccessFilter implements Filter {
    // 硬编码允许访问的IP列表
    private final Set<String> ALLOWED_IPS = new HashSet<>() {{
        add("192.168.1.100");
        add("192.168.1.101");
        add("127.0.0.1");
    }};

    @Override
    public void init(FilterConfig filterConfig) throws ServletException {}

    @Override
    public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException {
        String clientIp = request.getRemoteAddr();
        if (ALLOWED_IPS.contains(clientIp)) {
            // IP在允许列表中，继续处理请求
            chain.doFilter(request, response);
        } else {
            // IP不在列表中，返回403禁止访问
            ((HttpServletResponse) response).sendError(HttpServletResponse.SC_FORBIDDEN, "Access Denied: Your IP is not allowed");
        }
    }

    @Override
    public void destroy() {}
}

步骤2：配置SparkConf启用Filter #

在启动Worker前，将自定义Filter配置到SparkConf中，让Web UI加载这个过滤器：

// 初始化SparkConf
SparkConf conf = new SparkConf();

// 指定Web UI使用自定义IP过滤Filter（替换为你的Filter完整类名）
conf.set("spark.ui.filters", "com.yourpackage.IPAccessFilter");

2. RPC通信的IP过滤（基于Akka配置） #

Spark Worker的RPC通信基于Akka实现，你可以通过Akka的配置参数直接限制允许连接的IP地址：

在SparkConf中添加Akka的IP白名单配置，硬编码允许的RPC客户端IP：

// 硬编码允许的RPC客户端IP列表，支持逗号分隔和通配符（如192.168.1.*）
String allowedRpcIps = "192.168.1.100,192.168.1.101,127.0.0.1";

// 配置Akka只接受来自指定IP的RPC连接
conf.set("akka.remote.netty.tcp.allowed-addresses", allowedRpcIps);
// 可选：绑定Worker RPC到指定IP，进一步限制监听范围
conf.set("spark.worker.rpc.address", args.host());

3. 整合到你的启动代码 #

将上述配置整合到现有的Worker启动逻辑中：

public class CustomSparkWorker {
    public static void main(String[] args) {
        // 假设这里已经完成args参数的解析（host、port、webUiPort等）
        // ...

        SparkConf conf = new SparkConf();

        // 配置Web UI IP过滤
        conf.set("spark.ui.filters", "com.yourpackage.IPAccessFilter");

        // 配置RPC IP过滤
        String allowedRpcIps = "192.168.1.100,192.168.1.101,127.0.0.1";
        conf.set("akka.remote.netty.tcp.allowed-addresses", allowedRpcIps);

        // 启动Spark Worker
        Worker.startRpcEnvAndEndpoint(
            args.host(), 
            args.port(), 
            args.webUiPort(), 
            args.cores(), 
            args.memory(), 
            args.masters(), 
            args.workDir(), 
            scala.Option.empty(), 
            conf
        );
    }
}

注意事项 #

• 如果后续需要动态调整IP列表，建议将IP列表移到配置文件中，而非硬编码，便于维护。
• 如果Worker部署在反向代理之后，request.getRemoteAddr()获取的是代理服务器IP，此时需要从X-Forwarded-For请求头中提取真实客户端IP，修改Filter逻辑即可。
• Akka的allowed-addresses支持通配符格式（如192.168.1.*），可以简化IP范围的配置。

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