flink sql实现kafka落hdfs

Flink SQL是Apache Flink的一种查询语言，用于在Flink中进行实时数据处理和分析。要实现将Kafka中的数据落盘到HDFS，可以使用Flink SQL的相关功能。

首先，我们需要在Flink的配置文件中设置Kafka和HDFS的连接信息。在Flink的conf/flink-conf.yaml文件中，配置以下属性：

state.backend: filesystem
state.checkpoints.dir: hdfs://<HDFS_HOST>:<HDFS_PORT>/checkpoints
state.savepoints.dir: hdfs://<HDFS_HOST>:<HDFS_PORT>/savepoints

其中，<HDFS_HOST>是HDFS的主机地址，<HDFS_PORT>是HDFS的端口号。这样配置后，Flink将会将检查点和保存点存储到HDFS中。

接下来，在Flink SQL中创建一个表来读取Kafka中的数据，并将数据写入到HDFS中。可以使用以下SQL语句实现：

CREATE TABLE kafka_source (
  key STRING,
  value STRING
) WITH (
  'connector' = 'kafka',
  'topic' = '<KAFKA_TOPIC>',
  'properties.bootstrap.servers' = '<KAFKA_BOOTSTRAP_SERVERS>',
  'properties.group.id' = '<KAFKA_GROUP_ID>',
  'format' = 'json'
);

CREATE TABLE hdfs_sink (
  key STRING,
  value STRING
) WITH (
  'connector' = 'filesystem',
  'path' = 'hdfs://<HDFS_HOST>:<HDFS_PORT>/output',
  'format' = 'csv',
  'csv.field-delimiter' = ','
);

INSERT INTO hdfs_sink
SELECT key, value FROM kafka_source;

这里，'<KAFKA_TOPIC>'是Kafka中的主题名称，'<KAFKA_BOOTSTRAP_SERVERS>'是Kafka的启动服务器地址，'<KAFKA_GROUP_ID>'是Kafka消费者组的ID。'json'和'csv'是数据的格式，可以根据实际情况进行调整。

以上SQL语句创建了一个名为kafka_source的输入表，将Kafka中的数据源与之关联。同时，创建了一个名为hdfs_sink的输出表，将数据写入到HDFS中。最后，通过INSERT INTO语句，将kafka_source中的数据写入到hdfs_sink中。

通过以上的配置和操作，Flink SQL就可以实现将Kafka中的数据落盘到HDFS。