HDFS connector允许以各种格式将Kafka topic中的数据导出到HDFS文件中，并与Hive集成，使数据可以被HiveQL查询。
connector定期从Kafka轮询数据并将其写入HDFS。每个Kafka topic的数据由partitioner进行分区并划分为块。
每个数据块都表示为一个HDFS文件，其中包含topic、kafka partition，以及该数据块的开始和结束偏移量。如果配置中没有指定partitioner，则使用保留Kafka分区的默认partitioner。每个数据块的大小由写入HDFS的记录数量、写入时间以及schema兼容性决定。
HDFS connector可以和Hive集成，当启用时，connector自动为每个Kafka主题创建一个Hive外部分区表，并根据HDFS中可用数据更新表。
快速开始
1.从github下载kafka-connect-hdfs代码：
git clone https://github.com/confluentinc/kafka-connect-hdfs.git
2.编译
mvn clean install
3.进入target目录，将jar包拷到kafka目录的libs下
cp kafka-connect-hdfs-5.1.0.jar /opt/kafka_2.11-2.1.0/libs/
4.解决依赖问题，kafka-connect-hdfs-5.1.0.jar依赖别的jar包，这些jar包都在target目录下的kafka-connect-hdfs-5.1.0-development/share/java/kafka-connect-hdfs里，将依赖到的jar包也拷贝到kafka目录的libs下
5.启动worker
bin/connect-distributed.sh config/hdfs/connect-distributed-hdfs.properties
connect-distributed-hdfs.properties文件内容：
bootstrap.servers=master:9092
group.id=connect-cluster
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false
offset.storage.topic=connect-offsets-hdfs
offset.storage.replication.factor=1
offset.storage.partitions=1
config.storage.topic=connect-configs-hdfs
config.storage.replication.factor=1
config.storage.partitions=1
status.storage.topic=connect-status-hdfs
status.storage.replication.factor=1
status.storage.partitions=1
offset.flush.interval.ms=10000
plugin.path=/opt/kafka_2.11-2.1.0/libs,/opt/kafka-connector/kafka-connect-hdfs/target/kafka-connect-hdfs-5.1.0-development/share/java/kafka-connect-hdfs
hive.integration=true
hive.metastore.uris=thrift://master:9083/default
schema.compatibility=BACKWARD
6.启动connector
使用postman发送请求：
请求地址：http://slave1:8083/connectors
请求方式：POST
请求参数：
{
"name": "test-hive",
"config": {
"connector.class": "io.confluent.connect.hdfs.HdfsSinkConnector",
"tasks.max": "1",
"topics": "test_hive",
"hdfs.url": "hdfs://master:9000",
"flush.size": "3",
"name": "test-hive"
}
}
特性
HDFS connector提供以下特性：
- Exactly once：connector使用wal机制确保每条记录准确地导出到HDFS一次。此外，connector将Kafka的offsets保存在文件中来管理偏移量，以便在失败和任务重启时，可以从上次提交的偏移量开始。
- 可扩展的数据格式：connector原生支持Avro和parquet格式。此外，可以通过扩展Format类使用其他格式。
- Hive整合：connector支持开箱即用的Hive集成，启用时，connector自动为导出到HDFS的每个topic创建一个Hive外部分区表。
- Schema演化：connector支持schema演化以及不同的schema兼容性级别。当connector观察到schema改变时，它根据schema.compatibility配置选定正确的schema。
- 安全的HDFS和Hive Metastore支持：connector支持Kerberos身份验证，因此可以使用安全的HDFS和Hive metastore。
- 可插入的partitioner：connector支持默认的分区器、基于字段的分区器和基于事件的分区器。可以通过扩展Partitioner类来实现自己的分区器。此外，还可以通过扩展TimeBasedPartitioner类自定义基于时间的分区器。
Schema演化
HDFS connector支持schema演化，并根据schema.compatibility的配置对schema的更改做出响应。schema.compatibility支持四种模式：NONE,BACKWARD,FORWARD,ALL。
- NONE：这种情况下，connector确保写入HDFS的每个文件都具有适合的模式。当connector观察到数据中的schema更改时，它为受影响的topic分区提交当前文件集，并在新文件中使用新模式写入数据。
- BACKWARD：我们总是可以用最后的schema统一查询所有数据。例如，删除字段是对schema的向后兼容，因为当我们遇到使用包含这些字段的旧的schema写入的记录时，我们可以忽略它们。添加具有默认值的字段也是向后兼容的。
schema.compatibility设置为BACKWARD时，connector跟踪HDFS写入数据的最新schema，如果一个数据记录schema版本比当前版本新，connector提交当前文件，并用新schema写入文件。对于使用较早schema到达较晚的数据记录，connector将数据记录投射为最新的schema，然后再写入HDFS的同一组文件。 - FORWARD：如果schema.compatibility设置为FORWARD，我们总是使用最旧的schema统一查询数据。删除具有默认值的字段是向前兼容的，因为当字段丢失时，旧schema将使用默认值。
如果schema.compatibility设置为FORWARD，connector在写入HDFS中的同一组文件时，会将数据投射为最老的schema。 - FULL：完全兼容意味着可以用旧schema读取新数据，也可以用新schema读取旧数据。
如果schema.compatibility设置为FULL，connector表现的动作同BACKWORD相同
