FileSink 往hdfs上写入数据，怎么设置 Configuration的参数？找了好久 没找到

要在Apache Flink中使用FileSink将数据写入HDFS，并配置相关参数，您需要在Flink程序中设置Configuration对象并传递给FileSink。以下是如何进行配置的步骤概述：

1. 任务描述：

   • 配置Flink作业以通过FileSink将数据输出到HDFS上，涉及指定HDFS地址、文件格式等参数。

2. 前提条件：

   • 确保您的Flink环境已经正确配置了对Hadoop HDFS的支持，包括Hadoop依赖和相关环境变量。
   • 您拥有向目标HDFS目录写入数据的权限。

3. 具体步骤：

   import org.apache.flink.api.common.serialization.SimpleStringSchema;
   import org.apache.flink.configuration.Configuration;
   import org.apache.flink.connector.file.sink.FileSink;
   import org.apache.flink.core.fs.Path;
   import org.apache.flink.streaming.api.datastream.DataStream;
   import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
   
   public class WriteToHDFSExample {
      public static void main(String[] args) throws Exception {
          final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
   
          // 假设您有一个DataStream<String> dataStream，这里仅示例创建一个简单的流
          DataStream<String> dataStream = env.fromElements("record1", "record2", "record3");
   
          // 创建Configuration对象并设置HDFS相关的参数
          Configuration conf = new Configuration();
          conf.set("fs.defaultFS", "hdfs://your-hdfs-namenode:port"); // <sup>[1]</sup>
          // 可根据需要添加其他Hadoop配置，如HA、安全认证等
   
          // 配置FileSink
          FileSink<String> sink = FileSink
              .forRowFormat(new Path("hdfs://your-hdfs-path/your-directory"), new SimpleStringSchema())
              .withBucketAssigner(new DateTimeBucketAssigner<>("yyyy-MM-dd--HH")) // 示例：按时间分桶
              .withRollingPolicy(OnCheckpointRollingPolicy.build()) // 或其他滚动策略
              .withOutputFileConfig(FileSink.OutputFileConfig.builder()
                  .withPartPrefix("part-")
                  .withPartSuffix(".txt")
                  .build())
              .withFileSystemProperties(conf) // 将之前设置的HDFS配置传入
              .build();
   
          // 将DataStream连接到FileSink
          dataStream.sinkTo(sink);
   
          env.execute("Write to HDFS Example");
      }
   }
   

4. 注意事项：

   • 确保HDFS地址正确：请替换"hdfs://your-hdfs-namenode:port"和"hdfs://your-hdfs-path/your-directory"为实际的HDFS NameNode地址和目标目录路径。
   • 权限与安全性：如果HDFS集群启用了Kerberos等安全措施，还需在Flink配置中添加相应的安全设置。
   • 资源管理：根据实际情况调整文件滚动策略、压缩等配置，以优化存储空间和读取效率。

5. 验证与测试：

   • 运行上述Flink作业后，检查指定的HDFS目录是否成功接收到数据文件。
   • 使用HDFS命令行工具（如hadoop fs -ls /path/to/your/directory）或HDFS Web UI来验证数据写入情况。

通过以上步骤，您可以配置Flink的FileSink以将数据写入HDFS，并自定义相关参数满足特定需求。