如何将Spark Streaming的JavaDStream保存为.gz压缩文件？

嘿，刚接触Spark Streaming的话，要把输出改成.gz压缩文件其实并不难，我来一步步给你讲清楚怎么改～

首先，Spark Streaming底层依赖Hadoop的文件系统API，所以我们可以通过配置Hadoop的压缩参数来实现Gzip压缩输出。具体步骤如下：

1. 导入必要的压缩类 #

首先要导入Hadoop的GzipCodec类，这样我们才能指定用Gzip格式压缩：

import org.apache.hadoop.io.compress.GzipCodec;

2. 配置Hadoop输出压缩参数 #

在初始化你的StreamingContext之后，添加以下配置，开启压缩并指定Gzip编码：

// 假设你已经初始化了StreamingContext实例ssc
ssc.sparkContext().hadoopConfiguration().setBoolean("mapreduce.output.fileoutputformat.compress", true);
ssc.sparkContext().hadoopConfiguration().setClass(
    "mapreduce.output.fileoutputformat.compress.codec",
    GzipCodec.class,
    org.apache.hadoop.io.compress.CompressionCodec.class
);
// 设置压缩类型为块级压缩（比记录级压缩更高效）
ssc.sparkContext().hadoopConfiguration().set("mapreduce.output.fileoutputformat.compress.type", "BLOCK");

3. 修改保存文件的代码 #

最后，把你原来的saveAsTextFiles调用的后缀改成"gz"（这一步可选，但能让生成的文件名更清晰）：

// 替换原来的saveAsTextFiles调用
dataStreams.dstream().saveAsTextFiles(outputDir, "gz");

完整代码示例 #

把这些整合起来，你的代码大概会是这样：

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.hadoop.io.compress.GzipCodec;

public class GzipStreamingOutput {
    public static void main(String[] args) throws Exception {
        // 初始化Spark配置和StreamingContext
        SparkConf conf = new SparkConf().setAppName("GzipOutputDemo").setMaster("local[*]");
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(10));

        // 配置Gzip压缩
        jssc.sparkContext().hadoopConfiguration().setBoolean("mapreduce.output.fileoutputformat.compress", true);
        jssc.sparkContext().hadoopConfiguration().setClass(
            "mapreduce.output.fileoutputformat.compress.codec",
            GzipCodec.class,
            org.apache.hadoop.io.compress.CompressionCodec.class
        );
        jssc.sparkContext().hadoopConfiguration().set("mapreduce.output.fileoutputformat.compress.type", "BLOCK");

        // 你的DStream处理逻辑（这里用你原来的代码）
        JavaDStream<String> dataStreams = stream.map(new Function<String, String>() {
            @Override
            public String call(String lines) throws Exception {
                // 你的业务处理代码
                return lines;
            }
        });

        // 保存为Gzip压缩文件
        dataStreams.dstream().saveAsTextFiles("/path/to/output", "gz");

        jssc.start();
        jssc.awaitTermination();
    }
}

注意事项 #

• 确保你的项目依赖中包含了Hadoop的相关jar包，Spark 2.3.0默认已经集成了基础的压缩Codec，所以本地测试一般没问题；如果是集群环境，只要集群的Hadoop配置支持Gzip（大部分默认都支持）就不用额外配置。
• 当开启压缩后，Spark会自动将输出文件压缩为.gz格式，即使你不修改后缀，文件名也会自动带上.gz，但手动指定后缀"gz"会让文件名更直观。

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