flink sink hdfs

您好，您的问题是关于Flink中将数据写入HDFS的问题。对于这个问题，您可以使用Flink的HDFS Sink来实现将数据写入HDFS。具体步骤如下：

1. 在pom.xml中添加以下依赖：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-hadoop-compatibility_2.12</artifactId>
    <version>${flink.version}</version>
</dependency>

2. 在Flink代码中，使用以下代码创建HDFS Sink：

// 创建Hadoop Configuration对象
Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://localhost:9000");

// 创建HDFS Sink
Path outputPath = new Path("hdfs://localhost:9000/path/to/output");
TextOutputFormat<String> outputFormat = new TextOutputFormat<>(outputPath);
outputFormat.setWriteMode(WriteMode.OVERWRITE);
outputFormat.setOutputDirectoryMode(FileOutputFormat.OutputDirectoryMode.PARONLY);
HadoopFileSystem hdfs = new HadoopFileSystem(outputPath.toUri(), conf);
HadoopOutputFormat<Text, String> hadoopOutputFormat = new HadoopOutputFormat<>(outputFormat, conf);
hadoopOutputFormat.setExternalFileSystem(hdfs);
DataStream<String> dataStream = ... // 输入数据流
dataStream.writeUsingOutputFormat(hadoopOutputFormat);

其中，`outputPath`是输出路径，`TextOutputFormat`是输出格式，`WriteMode.OVERWRITE`表示覆盖已有文件，`FileOutputFormat.OutputDirectoryMode.PARONLY`表示只输出到HDFS的partition目录，`HadoopFileSystem`是Hadoop的FileSystem对象，`HadoopOutputFormat`是Hadoop的OutputFormat对象，`dataStream`是输入数据流。

这样，您就可以通过Flink将数据写入HDFS了。希望对您有所帮助。