代码之家 › 专栏 › 技术社区 › Yuriy Bondaruk

具有多个加密密钥提供程序的EMR

amazon-emr emr spark-streaming encryption apache-spark

Yuriy Bondaruk · 技术社区 · 8 年前

我正在运行启用的EMR群集 s3 client-side encryption 使用自定义密钥提供程序。但现在我需要使用不同的加密模式将数据写入多个s3目标:

CSE自定义密钥提供程序
CSE-KMS

通过定义s3 bucket和加密类型之间的某种映射,是否可以将EMR配置为使用这两种加密类型?

或者,由于我使用spark结构化流来处理数据并将数据写入s3,我想知道是否可以禁用EMRFS上的加密,然后分别为每个流启用CSE?

3 回复 | 直到 8 年前

Yuriy Bondaruk 8 年前

其思想是支持任何文件系统方案并对其进行单独配置。例如:

# custom encryption key provider
fs.s3x.cse.enabled = true
fs.s3x.cse.materialsDescription.enabled = true
fs.s3x.cse.encryptionMaterialsProvider = my.company.fs.encryption.CustomKeyProvider

#no encryption
fs.s3u.cse.enabled = false

#AWS KMS
fs.s3k.cse.enabled = true
fs.s3k.cse.encryptionMaterialsProvider = com.amazon.ws.emr.hadoop.fs.cse.KMSEncryptionMaterialsProvider
fs.s3k.cse.kms.keyId = some-kms-id

然后在spark中使用它,如下所示:

StreamingQuery writeStream = session
        .readStream()
        .schema(RecordSchema.fromClass(TestRecord.class))
        .option(OPTION_KEY_DELIMITER, OPTION_VALUE_DELIMITER_TAB)
        .option(OPTION_KEY_QUOTE, OPTION_VALUE_QUOTATION_OFF)
        .csv(âs3x://aws-s3-bucket/inputâ)
        .as(Encoders.bean(TestRecord.class))
        .writeStream()
        .outputMode(OutputMode.Append())
        .format("parquet")
        .option("path", âs3k://aws-s3-bucket/outputâ)
        .option("checkpointLocation", âs3u://aws-s3-bucket/checkpointingâ)
        .start();

Ta处理此问题我实现了一个自定义Hadoop文件系统(扩展 org.apache.hadoop.fs.FileSystem )它将调用委托给真实的文件系统,但需要修改配置。

// Create delegate FS
this.config.set("fs.s3n.impl", âcom.amazon.ws.emr.hadoop.fs.EmrFileSystemâ);
this.config.set("fs.s3n.impl.disable.cache", Boolean.toString(true));
this.delegatingFs = FileSystem.get(s3nURI(originalUri, SCHEME_S3N), substituteS3Config(conf));

传递给委派文件系统的配置应采用所有原始设置,并替换任何出现的 fs.s3*. 具有 fs.s3n. .

private Configuration substituteS3Config(final Configuration conf) {
    if (conf == null) return null;

    final String fsSchemaPrefix = "fs." + getScheme() + ".";
    final String fsS3SchemaPrefix = "fs.s3.";
    final String fsSchemaImpl = "fs." + getScheme() + ".impl";
    Configuration substitutedConfig = new Configuration(conf);
    for (Map.Entry<String, String> configEntry : conf) {
        String propName = configEntry.getKey();
        if (!fsSchemaImpl.equals(propName)
            && propName.startsWith(fsSchemaPrefix)) {
            final String newPropName = propName.replace(fsSchemaPrefix, fsS3SchemaPrefix);
            LOG.info("Substituting property '{}' with '{}'", propName, newPropName);
            substitutedConfig.set(newPropName, configEntry.getValue());
        }
    }

    return substitutedConfig;
}

除此之外,请确保委派fs接收带有支持方案的URI和路径,并返回带有自定义方案的路径

@Override
public FileStatus getFileStatus(final Path f) throws IOException {
    FileStatus status = this.delegatingFs.getFileStatus(s3Path(f));
    if (status != null) {
        status.setPath(customS3Path(status.getPath()));
    }
    return status;
}

private Path s3Path(final Path p) {
    if (p.toUri() != null && getScheme().equals(p.toUri().getScheme())) {
        return new Path(s3nURI(p.toUri(), SCHEME_S3N));
    }
    return p;
}

private Path customS3Path(final Path p) {
    if (p.toUri() != null && !getScheme().equals(p.toUri().getScheme())) {
        return new Path(s3nURI(p.toUri(), getScheme()));
    }
    return p;
}

private URI s3nURI(final URI originalUri, final String newScheme) {
     try {
         return new URI(
             newScheme,
             originalUri.getUserInfo(),
             originalUri.getHost(),
             originalUri.getPort(),
             originalUri.getPath(),
             originalUri.getQuery(),
             originalUri.getFragment());
     } catch (URISyntaxException e) {
         LOG.warn("Unable to convert URI {} to {} scheme", originalUri, newScheme);
     }

     return originalUri;
}

最后一步是用Hadoop注册自定义文件系统( spark-defaults 分类)

spark.hadoop.fs.s3x.impl = my.company.fs.DynamicS3FileSystem
spark.hadoop.fs.s3u.impl = my.company.fs.DynamicS3FileSystem
spark.hadoop.fs.s3k.impl = my.company.fs.DynamicS3FileSystem

stevel 8 年前

我不能代表Amazon EMR,但在hadoop的s3a连接器上,可以逐个设置加密策略。然而,S3A不支持客户端加密,因为它打破了关于文件长度的基本假设(您可以读取的数据量必须==目录列表/getFileStatus调用中的长度)。

我希望亚马逊也会做类似的事情。您可以创建自定义Hadoop Configuration 具有不同设置的对象(&A);使用它来检索用于保存内容的文件系统实例。不过,Spark很棘手。

Sacha Epskamp 5 年前

使用EMRFS时,可以按以下格式指定每个bucket的配置:

fs.s3.bucket.<bucket name>.<some.configuration>

例如,要关闭CSE(bucket除外) s3://foobar ,可以设置:

   "Classification": "emrfs-site",
   "Properties": {
      "fs.s3.cse.enabled": "false",
      "fs.s3.bucket.foobar.cse.enabled": "true",
      [your other configs as usual]
   }

请注意,它必须是 fs.s3 而不是 fs.{arbitrary-scheme} 喜欢 fs.s3n .