带密码保护的 Spark 数据帧输出压缩（gzip）答案

【问题标题】：Spark dataframe output compression(gzip) with password protected带密码保护的 Spark 数据帧输出压缩（gzip）
【发布时间】：2020-12-29 15:05:54
【问题描述】：

使用下面的代码，我可以将其压缩并保存为 .gz 文件

import spark.implicits._
 

val someDF = Seq(
  (8, "bat"),
  (64, "mouse"),
  (-27, "horse")
).toDF("number", "word")

someDF.coalesce(1)
   .write.format("com.databricks.spark.csv")
    .option("header", "true")
    .option("codec", "org.apache.hadoop.io.compress.GzipCodec")
    .save("example.csv.gz")

spark 是否提供了一个选项来压缩受密码保护的数据？我无法在 spark 文档中找到。

【问题讨论】：

标签： scala apache-spark apache-spark-sql spark-streaming hadoop2

【解决方案1】：

可以创建先压缩文件然后加密它们的新编解码器。这个想法是在写入文件系统之前用CipherOutputStream 包装编解码器的输出流。

import java.io.{IOException, OutputStream}

import javax.crypto.{Cipher, CipherOutputStream}
import javax.crypto.spec.SecretKeySpec
import org.apache.hadoop.io.compress._


class GzipEncryptionCodec extends GzipCodec {

  override def getDefaultExtension(): String = ".gz.enc"

  @throws[IOException]
  override def createOutputStream(out: OutputStream): CompressionOutputStream =
    super.createOutputStream(wrapWithCipherStream(out))

  @throws[IOException]
  override def createOutputStream(out: OutputStream, compressor: Compressor): CompressionOutputStream =
    super.createOutputStream(wrapWithCipherStream(out), compressor)

  def wrapWithCipherStream(out: OutputStream): OutputStream = {
    val cipher = Cipher.getInstance("AES/ECB/PKCS5Padding") //or another algorithm
    val secretKey = new SecretKeySpec(
      "hello world 1234".getBytes, //this is not a secure password!
      "AES")
    cipher.init(Cipher.ENCRYPT_MODE, secretKey)
    return new CipherOutputStream(out, cipher)
  }
}

在编写 csv 文件时，可以使用此编解码器：

df.write
  .option("codec","GzipEncryptionCodec")
  .mode(SaveMode.Overwrite).csv("encryped_csv")

并且输出文件将被加密并得到后缀.gz.enc。

此编解码器仅加密数据，无法解密。关于为什么更改编解码器以读取比写入更困难的一些背景知识可以在here 找到。

相反，可以使用简单的 Scala 程序读取和解密文件：

import javax.crypto.Cipher
import javax.crypto.spec.SecretKeySpec
import java.io.FileInputStream
import java.util.zip.GZIPInputStream

import javax.crypto.CipherInputStream
val cipher = Cipher.getInstance("AES/ECB/PKCS5Padding")
val secretKey = new SecretKeySpec("hello world 1234".getBytes(), "AES")
cipher.init(Cipher.DECRYPT_MODE, secretKey)

val files = new File("encryped_csv").listFiles.filter(_.getName().endsWith(".gz.enc")).toList

files.foreach(f => {
  val dec = new CipherInputStream(new FileInputStream(f), cipher)
  val gz = new GZIPInputStream(dec)
  val result = scala.io.Source.fromInputStream(gz).mkString
  println(f.getName)
  println(result)
})

【讨论】：

【解决方案2】：

Gzip 本身不支持密码保护。在 Unix 上，您需要使用其他工具来使用密码对文件进行加密。

附：此外，将 com.databricks.spark.csv 替换为 csv - Spark 已经支持 CSV 很长时间了。并移除对应的Maven依赖。

【讨论】：

任何其他 spark 编解码器/lib 提供压缩 csv 并受密码保护的能力？
这个库 (github.com/srikanth-lingala/zip4j) 可以帮助我猜我必须将所有 csv 文件保存在本地，然后创建一个带有密码保护的 zip。但我不确定这是否可以在集群模式下工作。
我不知道有什么东西可以用于 CSV + Spark 中的东西组合。我认为您可能需要求助于低级的东西，例如 forEachRDD，“手动”生成 CSV 并自己编写文件。但如果您对 Spark 不是很深入，这将是一个挑战