Flink sql方式可以直接写HDFS吗？不要通过Hive实现，有的帮忙发个demo，谢了？

Flink SQL 可以直接写入 HDFS，而无需通过 Hive 实现。以下是实现方式和一个示例代码。

前提条件

1. 已创建 EMR 集群：需要确保集群版本为 EMR-3.42.0 及以上或 EMR-5.8.0 及以上。
2. 已开通并授权访问 OSS-HDFS 服务：如果目标存储是阿里云 OSS 的 HDFS 服务，需完成相关配置。
3. Flink 环境已正确配置 HDFS Connector：确保 Flink 能够识别 HDFS 的路径和存储格式。

实现步骤

1. 配置 HDFS 相关参数

在 Flink 作业中，可以通过 WITH 子句指定 HDFS 的连接信息。例如： - connector：指定为 filesystem。 - path：指定 HDFS 文件路径。 - format：指定文件格式（如 CSV、JSON、Parquet 等）。

2. 编写 Flink SQL 示例代码

以下是一个将数据写入 HDFS 的完整示例：

-- 创建源表（假设从 Kafka 中读取数据）
CREATE TABLE source_table (
    id INT,
    name STRING,
    age INT
) WITH (
    'connector' = 'kafka',
    'topic' = 'input_topic',
    'properties.bootstrap.servers' = 'localhost:9092',
    'format' = 'json'
);

-- 创建目标表（写入 HDFS）
CREATE TABLE sink_table (
    id INT,
    name STRING,
    age INT
) WITH (
    'connector' = 'filesystem',
    'path' = 'hdfs://<namenode-host>:<port>/user/flink/output', -- 替换为实际 HDFS 路径
    'format' = 'csv' -- 指定文件格式
);

-- 插入数据到目标表
INSERT INTO sink_table
SELECT id, name, age
FROM source_table;

参数说明

1. connector：指定为 filesystem，表示使用文件系统作为存储介质。
2. path：HDFS 的路径，需替换为实际的 NameNode 地址和端口。
3. format：支持多种文件格式，如 csv、json、parquet 等，根据需求选择。

注意事项

• HDFS 配置：确保 Flink 集群能够访问 HDFS，并正确配置 core-site.xml 和 hdfs-site.xml 文件。
• 权限问题：如果 HDFS 启用了 Kerberos 认证，需在 Flink 集群中配置相应的认证信息。
• 性能优化：对于大规模数据写入，建议启用 Flink 的 Checkpoint 功能以保证数据一致性。

扩展功能

如果需要写入 OSS-HDFS 服务，可以将 path 替换为 oss:// 前缀的路径，并配置 OSS 的 Endpoint 和访问密钥。例如：

'path' = 'oss://<bucket-name>.<oss-hdfs-endpoint>/user/flink/output',
'format' = 'csv',
'fs.oss.accessKeyId' = '<your-access-key-id>',
'fs.oss.accessKeySecret' = '<your-access-key-secret>',
'fs.oss.endpoint' = '<your-oss-endpoint>'

通过上述方法，您可以直接使用 Flink SQL 将数据写入 HDFS 或 OSS-HDFS，而无需依赖 Hive。