流程流程:
Flink消费Kafka,逻辑处理后将实时流转换为表视图,利用HiveCataLog创建Hive表,将实时流 表insert进Hive,注意分区时间字段需要为 yyyy-MM-dd形式,否则抛出异常:java.time.format.DateTimeParseException: Text '20240111' could not be parsed
写入到hive分区表
- streamEnv需要开启checkpoint,保证flink写入hive分区表的写入一致性
- hive表ddl中需要指定以下TBLPROPERTIES:
- sink.partition-commit.trigger:分区提交触发器,单选,可选值为partition-time、process-time(默认), 其中==partition-time需要根据当前数据的watermark来判断分区是否需要提交,当watermark + delay大于等于分区上的时间时就会提交该分区元数据==;process-time的话根据当前系统处理时间来判断分区是否需要提交,当系统处理时间大于等于分区上的时间就会提交该分区元数据
- partition.time-extractor.timestamp-pattern:使用partition-time触发器时使用该配置项。表示从表字段中提取出表达某个分区的时间的格式,==需要提取到的时间必须为yyyy-MM-dd HH:mm:ss的格式==。比如字段dt的格式为yyyy-MM-dd,则配置为$dt 00:00:00则表示分区时间取值为dt的value的0点0分0秒,可以选择多个表字段组合。当表字段无法抽取出符合的格式时,则使用自定义提取器partition.time-extractor.class。
- sink.partition-commit.delay: 表示watermark允许event time的最大乱序时间,使用partition-time触发器时可以使用,默认为0s
- sink.partition-commit.policy.kind:分区提交方式,多选,可选值为metastore、success-file、custom,metastore表示写入元数据库,success-file表示往hdfs分区目录写入一个标志文件,custom表示使用自定义提交方式,通常使用metastore,success-file组合
- partition.time-extractor.kind:当要使用自定义分区时间提取器时需要配置此项,值配置为custom
- partition.time-extractor.class:当要使用自定义分区时间提取器时需要配置此项,值配置为自定义提取器的类路径。在集群中运行时,需要把该类打成jar包放到flink lib目录下。
- 某个分区触发提交后,后续再有此分区的数据进来,仍然会写入hive该分区。
写入到hive非分区表
val streamEnv = ... val dataStream = ... val streamTableEnv = ... streamTableEnv.createTemporaryView("自定义catalog表名", dataStream, *fields) # 当前flink存在bug,转换时必须指定fields或者schema,否则watermark无法流入table val catalog = ... streamTableEnv.registerCatalog("hive", catalog) streamTableEnv.useCatalog("hive") streamTableEnv.executeSql("insert sql").print()
参考:
flink - sink - hive - 简书 (jianshu.com)
5-flinkSQL参数 (gitee.io)
详见官网:文章来源:https://www.toymoban.com/news/detail-793244.html
Catalogs | Apache Flink文章来源地址https://www.toymoban.com/news/detail-793244.html
到了这里,关于Flink集成Hive之Hive Catalog的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!