Spark SASL 无法使用纱线处理 emr

Question

首先，我想说我看到的唯一解决此问题的方法是：Spark 1.6.1 SASL。但是，在添加 spark 和 yarn 身份验证的配置时，仍然无法正常工作。下面是我在亚马逊 emr 上的纱线集群上使用 spark-submit 的 spark 配置：

    SparkConf sparkConf = new SparkConf().setAppName("secure-test");
    sparkConf.set("spark.authenticate.enableSaslEncryption", "true");
    sparkConf.set("spark.network.sasl.serverAlwaysEncrypt", "true");
    sparkConf.set("spark.authenticate", "true");
    sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
    sparkConf.set("spark.kryo.registrator", "org.nd4j.Nd4jRegistrator");
    try {
        sparkConf.registerKryoClasses(new Class<?>[]{
                Class.forName("org.apache.hadoop.io.LongWritable"),
                Class.forName("org.apache.hadoop.io.Text")
        });
    } catch (Exception e) {}

    sparkContext = new JavaSparkContext(sparkConf);
    sparkContext.hadoopConfiguration().set("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem");
    sparkContext.hadoopConfiguration().set("fs.s3a.enableServerSideEncryption", "true");
    sparkContext.hadoopConfiguration().set("spark.authenticate", "true");

注意，我在代码中将 spark.authenticate 添加到 sparkContext 的 hadoop 配置中，而不是 core-site.xml（我假设我可以这样做，因为其他东西也可以）。

看这里：https://github.com/apache/spark/blob/master/common/network-yarn/src/main/java/org/apache/spark/network/yarn/YarnShuffleService.java 似乎 spark.authenticate 都是必要的。当我运行这个应用程序时，我得到以下堆栈跟踪。

17/01/03 22:10:23 INFO storage.BlockManager: 向本地外部 shuffle 服务注册执行程序。 17/01/03 22:10:23 错误 client.TransportClientFactory：178 毫秒后引导客户端时出现异常 java.lang.RuntimeException：java.lang.IllegalArgumentException：未知消息类型：-22 在 org.apache.spark.network.shuffle.protocol.BlockTransferMessage$Decoder.fromByteBuffer(BlockTransferMessage.java:67) 在 org.apache.spark.network.shuffle.ExternalShuffleBlockHandler.receive（ExternalShuffleBlockHandler.java:71） 在 org.apache.spark.network.server.TransportRequestHandler.processRpcRequest(TransportRequestHandler.java:149) 在 org.apache.spark.network.server.TransportRequestHandler.handle(TransportRequestHandler.java:102) 在 org.apache.spark.network.server.TransportChannelHandler.channelRead0(TransportChannelHandler.java:104) 在 org.apache.spark.network.server.TransportChannelHandler.channelRead0(TransportChannelHandler.java:51) 在 io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105) 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:333) 在 io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:319) 在 io.netty.handler.timeout.IdleStateHandler.channelRead(IdleStateHandler.java:254) 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:333) 在 io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:319) 在 io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103) 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:333) 在 io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:319) 在 org.apache.spark.network.util.TransportFrameDecoder.channelRead(TransportFrameDecoder.java:86) 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:333) 在 io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:319) 在 io.netty.channel.DefaultChannelPipeline.fireChannelRead（DefaultChannelPipeline.java:787） 在 io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:130) 在 io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511) 在 io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized (NioEventLoop.java:468) 在 io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382) 在 io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354) 在 io.netty.util.concurrent.SingleThreadEventExecutor$2.run（SingleThreadEventExecutor.java:116） 在 java.lang.Thread.run(Thread.java:745)

在 Spark 的文档中，它说

For Spark on YARN deployments, configuring spark.authenticate to true will automatically handle generating and distributing the shared secret. Each application will use a unique shared secret.

根据上面 yarn 文件中的 cmets，这似乎是错误的，但是通过故障排除，我仍然不知道应该去哪里让 sasl 工作？我是否遗漏了某处记录的明显内容？

Answer 1

所以我终于想通了。之前的 StackOverflow 线程在技术上是正确的。我需要将 spark.authenticate 添加到纱线配置中。也许可以做到这一点，但我不知道如何在代码中添加这个配置，这在高层次上是有道理的，为什么会这样。我将在下面发布我的配置，以防其他人将来遇到此问题。

首先，我使用了一个 aws emr 配置文件（例如，使用 aws cli aws emr create-cluster --configurations file://youpathhere.json 时）

然后，我将以下 json 添加到文件中：

[{
    "Classification": "spark-defaults",
    "Properties": {
        "spark.authenticate": "true",
        "spark.authenticate.enableSaslEncryption": "true",
        "spark.network.sasl.serverAlwaysEncrypt": "true"
    }
},
{
    "Classification": "core-site",
    "Properties": {
        "spark.authenticate": "true"
    }
}]

Answer 2

添加 Spark 网络加密的配置选项后，我在 Dataproc (Google Cloud Platform) 上的 Spark 上收到了相同的错误消息。

我最初使用以下命令创建了 Dataproc 集群。

gcloud dataproc clusters create test-encryption --no-address \
--service-account=<SERVICE-ACCOUNT> \
--zone=europe-west3-c --region=europe-west3 \
--subnet=<SUBNET> \
--properties 'spark:spark.authenticate=true,spark:spark.network.crypto.enabled=true'

解决方案是另外添加配置'yarn:spark.authenticate=true'。因此，可以按如下方式创建具有 Spark 的 RPC 加密的工作 Dataproc 集群。

gcloud dataproc clusters create test-encryption --no-address \
--service-account=<SERVICE-ACCOUNT> \
--zone=europe-west3-c --region=europe-west3 \
--subnet=<SUBNET> \
--properties 'spark:spark.authenticate=true,spark:spark.network.crypto.enabled=true,yarn:spark.authenticate=true'

我使用 ngrep 验证了加密。我在master节点上安装ngrep如下。

sudo apt-get update
sudo apt-get install ngrep

然后我在任意端口 20001 上运行 ngrep。

sudo ngrep port 20001

如果您随后使用以下配置属性运行 Spark 作业，您可以看到驱动程序和工作程序节点之间的加密通信。

spark.driver.port=20001
spark.blockManager.port=20002

注意，我总是建议在 Dataproc 上启用 Kerberos 以保护 Hadoop、Yarn 等的身份验证。这可以通过集群创建命令中的标志 --enable-kerberos 来实现。