flink写文件到hdfs，并指定gz压缩格式

要将文件写入HDFS并指定Gzip压缩格式，可以使用Flink提供的`org.apache.flink.core.fs.FileSystem`和`org.apache.flink.core.fs.Path`类来实现。具体步骤如下：

1. 创建一个`org.apache.flink.core.fs.FileSystem`对象，指定HDFS的URI和配置信息。

2. 创建一个`org.apache.flink.core.fs.Path`对象，指定写入HDFS的文件路径。

3. 调用`FileSystem.create()`方法创建一个输出流。

4. 将数据写入输出流，这里可以使用`org.apache.flink.api.common.io.FileOutputFormat`类来实现Gzip压缩。

5. 关闭输出流。

下面是一个示例程序，它将数据写入HDFS并使用Gzip压缩：

import org.apache.flink.api.common.io.FileOutputFormat;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class WriteToHdfsExample {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 准备要写入HDFS的数据
        DataStream<String> data = ...

        // 写入HDFS
        String outputPath = "hdfs://hadoop-master:9000/path/to/output";
        FileSystem hdfs = FileSystem.get(new URI(outputPath), new Configuration());
        Path path = new Path(outputPath);
        FSDataOutputStream outputStream = hdfs.create(path);
        GzipCompressor gzipCompressor = new GzipCompressor();
        FileOutputFormat<String> fileOutputFormat = new TextOutputFormat<>(path, gzipCompressor);
        fileOutputFormat.setOutputFilePath(path);
        fileOutputFormat.setWriteMode(FileSystem.WriteMode.OVERWRITE);
        fileOutputFormat.open(outputStream);
        data.writeUsingOutputFormat(fileOutputFormat);
        fileOutputFormat.close();

        // 启动任务执行
        env.execute("Write to HDFS Example");
    }
}

在以上示例程序中，`hadoop-master:9000`是HDFS的URI，`/path/to/output`是要写入的文件路径。`TextOutputFormat`是Flink提供的一个文本输出格式，它支持Gzip压缩。在`FileOutputFormat`的构造函数中，将`TextOutputFormat`作为参数传入，即可实现Gzip压缩。