flink如何写入数据到hdfs

Apache Flink 中，将数据写入 HDFS (Hadoop Distributed File System) 可以通过 Flink 的 `FlinkStream API` 或者 `FlinkTable API` 来完成，通常涉及到 `SinkFunction` 或 `DataStream.write()` 函数。以下是基本步骤：

1. **设置环境**：
首先，需要在 Flink 程序开始时配置 Hadoop FileSystem。添加 Hadoop的相关依赖，例如：

   Configuration conf = new Configuration();
   conf.set("fs.defaultFS", "hdfs://your-hdfs-url");
   Environment env = StreamExecutionEnvironment.getExecutionEnvironment();
   env.getConfig().setParallelism(1); // 根据实际情况调整并行度
   env.addConfig(conf);

2. **创建 DataStream 或 Table**：
如果是 `DataStream`，可以这样做：

   DataStream<String> text = ...; // 从源头获取数据流

对于 `Table` API，使用类似语法：

   Table table = ...;

3. **写入操作**：
使用 `DataStream.write()` 方法将数据写入文件系统，如：

   text.write()
       .format(new TextFileOutputFormat<>(
           "hdfs://path/to/your/output/directory",
           ","
       ))
       .disableSchema蒸馏()
       .finish();

对于 `Table` API，可以用 `toRetractStream()` 和 `executeInsertInto()`，比如：

   table.toRetractStream()
       .insertInto("your-hdfs-table-or-directory");

4. **提交任务**：
完成所有写操作后，提交作业到 Flink 环境运行：

   env.execute("Flink Write to HDFS Job");

记得替换 `'your-hdfs-url'`, `'your-output-directory'`, `'your-hdfs-table-or-directory'` 为实际的 HDFS URL 和路径。