mqtt 写入hive

### 将 MQTT 消息写入 Hive 的方法

为了实现将 MQTT 消息写入 Hive 数据库，可以采用多种技术栈组合来完成这一目标。一种推荐的方式是利用 Kafka Connect for MQTT 来桥接 MQTT 和 Kafka 之间的消息传递，再通过 Kafka Connect 插件将数据最终导入到 Hive 中[^1]。

#### 架构概述

架构设计如下：

1. **MQTT 到 Kafka**：使用 `Kafka Connect for MQTT` 连接器订阅来自特定主题的消息并将其转发给 Apache Kafka 集群。
2. **Kafka 到 Hive**：部署合适的 Kafka Connector (如 Confluent 提供的 HDFS Sink Connector)，配置其指向 Hive 表结构以及存储位置，在此过程中可能还需要设置 SerDe 序列化/反序列化工具以便正确解析 JSON 或其他格式的数据。

#### 实现步骤详解

##### 安装与配置 Kafka Connect for MQTT

确保已经安装好 Apache Kafka 及其相关组件之后，下载适用于环境版本的 `kafka-connect-mqtt` 并按照官方文档说明进行必要的参数调整以匹配本地网络状况和安全策略需求。

# 下载 kafka-connect-mqtt jar 文件至插件目录下
wget https://example.com/path/to/kafka-connect-mqtt.jar -P /path/to/plugins/

编辑 connector 属性文件 (`mqtt-source.properties`) 设置源端点信息:

name=mqtt-source
connector.class=io.confluent.connect.mqtt.MqttSourceConnector
tasks.max=1
mqtt.server.uri=tcp://localhost:1883
mqtt.topics=<your_topic>
kafka.topic=<target_kafka_topic>
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false

启动该连接器实例并将上述属性应用上去即可开始接收来自指定 MQTT 主题的新事件流。

##### 使用 HDFS/Hive Sink Connector 处理 Kafka 数据

接下来定义一个用于消费这些经过转换后的记录并向下游系统推送的目标 sink 接口；这里选用的是专门针对分布式文件系统的 HDFS Sink Connector ，它能够很好地配合 Hive Metastore 工作从而简化 ETL 流程管理。

创建一个新的 properties 文件名为 `hive-sink.properties`, 添加下面的内容：

name=hdfs-sink
connector.class=io.confluent.connect.hdfs.HdfsSinkConnector
tasks.max=1
topics=<target_kafka_topic>

hadoop.conf.dir=/etc/hadoop/conf
hive.integration=true
hive.database=default
hive.metastore.uris=thrift://<metastore_host>:9083
flush.size=100
rotate.interval.ms=-1
schema.compatibility=BACKWARD_COMPATIBLE
partitioner.class=io.confluent.connect.hdfs.partitioner.FieldPartitioner
partition.field.name=time_field_name
locale=en-US
timezone=UTC
format.class=io.confluent.connect.hdfs.parquet.ParquetFormat

注意替换 `<target_kafka_topic>`、`<metastore_host>` 等占位符为实际值，并根据业务逻辑自定义分区字段名称及其他高级选项。

最后一步是在命令行工具里提交这个新的任务描述：

curl -X POST \
     -H "Content-Type: application/json" \
     --data '{"name": "hive-sink", "config": '"$(cat hive-sink.properties | jq -Rs .)"'}' \
     http://localhost:8083/connectors

此时整个管道应该已经开始运作起来，每当有新条目进入原始 MQTT topic 后都会被自动捕获并通过一系列中间环节最终持久化保存到了 Hive 表内等待进一步分析查询操作。