网创优客建站品牌官网
为成都网站建设公司企业提供高品质网站建设
热线:028-86922220
成都专业网站建设公司

定制建站费用3500元

符合中小企业对网站设计、功能常规化式的企业展示型网站建设

成都品牌网站建设

品牌网站建设费用6000元

本套餐主要针对企业品牌型网站、中高端设计、前端互动体验...

成都商城网站建设

商城网站建设费用8000元

商城网站建设因基本功能的需求不同费用上面也有很大的差别...

成都微信网站建设

手机微信网站建站3000元

手机微信网站开发、微信官网、微信商城网站...

建站知识

当前位置:首页 > 建站知识

FlinkSQL如何连接Hive并写入/读取数据

这篇文章主要介绍Flink SQL如何连接Hive并写入/读取数据,文中介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们一定要看完!

成都创新互联公司是一家专业提供广德企业网站建设,专注与成都网站建设、成都网站制作、HTML5建站、小程序制作等业务。10年已为广德众多企业、政府机构等服务。创新互联专业网站设计公司优惠进行中。

1. 添加依赖

    
        1.11.2
        2.11
    

    
        
            org.apache.flink
            flink-streaming-scala_${scala.version}
            ${flink.version}
        
        
            org.apache.flink
            flink-connector-kafka-0.11_${scala.version}
            ${flink.version}
        
        
            org.apache.flink
            flink-clients_${scala.version}
            ${flink.version}
        

        
            org.apache.flink
            flink-table-api-java-bridge_${scala.version}
            ${flink.version}
        

        
            org.apache.flink
            flink-table-planner-blink_${scala.version}
            ${flink.version}
        

        
        
            org.apache.flink
            flink-connector-hive_${scala.version}
            ${flink.version}
        

        
        
            org.apache.hive
            hive-exec
            2.1.1
        

        
            org.apache.flink
            flink-shaded-hadoop-2-uber
            2.6.5-7.0
        

        
            org.apache.flink
            flink-json
            ${flink.version}
        

        
            org.apache.flink
            flink-connector-elasticsearch7_${scala.version}
            ${flink.version}
        

        
            org.apache.flink
            flink-csv
            ${flink.version}
        

        
            com.fasterxml.jackson.core
            jackson-databind
            2.4.0
        

        
            com.fasterxml.jackson.core
            jackson-annotations
            2.4.0
        

        
            com.fasterxml.jackson.core
            jackson-core
            2.4.0
        
    

2. 创建blink版本的批处理Table执行环境

EnvironmentSettings bbSettings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inBatchMode()
                .build();
TableEnvironment bbTableEnv = TableEnvironment.create(bbSettings);
  • 经过实际测试,目前HiveTableSink 不支持流式写入(未实现 AppendStreamTableSink),必须是批处理环境才可以往hive里面写入数据,而不能将流式数据写入hive。例如将kafka创建一张临时表,然后将表中的数据流持续插入hive,这是不可以的,官网上1.11版本通过flink sql-client可以实现hive的流式写入,还有待验证。

3. 连接文件系统,创建hive catalog,对表进行操作,类似于Spark on Hive,flink可以直接获取Hive的元数据,并使用flink进行计算。

        // 连接外部文件
        bbTableEnv.connect(new FileSystem().path("file:///E:/d.txt"))
                .withFormat(new Csv().fieldDelimiter(','))
                .withSchema(new Schema().field("id", DataTypes.STRING()))
                .createTemporaryTable("output");

        // 设置 hive 方言
        bbTableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
        // 获取hive-site.xml目录
        String hiveConfDir = Thread.currentThread().getContextClassLoader().getResource("").getPath().substring(1);
        HiveCatalog hive = new HiveCatalog("hive", "warningplatform", hiveConfDir);
        bbTableEnv.registerCatalog("hive", hive);

        bbTableEnv.useCatalog("hive");
        bbTableEnv.useDatabase("warningplatform");

        bbTableEnv.executeSql("insert into  test select id from    default_catalog.default_database.output");
  • 通过bbTableEnv.connect()去创建临时表的方式已经过时了,建议使用bbTableEnv.executeSql()的方式,通过DDL去创建临时表,临时表到底是属于哪一个catalog目前还不太确定,到底是什么规则目前还不清楚。 查资料得知,临时表与单个Flink会话的生命周期相关,临时表始终存储在内存中。 永久表需要一个catalog来管理表对应的元数据,比如hive metastore,该表将一直存在,直到明确删除该表为止。 因此猜测:default_catalog是存储在内存中,如果在切换成hive catalog之前创建临时表,那我们就可以使用default_catalog.default_database.tableName来获取这个临时表。 如果切换了catalog再去创建临时表,那我们就无法获取到临时表了,因为它不在default_catalog中,而且保存在内存里面,直接查询临时表会去当前的catalog里面去查找临时表,因此一定要在default_catalog 里面创建临时表。 而临时视图好像是存储在当前的catalog里面

  • 通过bbTableEnv.createTemporaryView()创建的视图则是属于当前的database的

    bbTableEnv.createTemporaryView("output",bbTableEnv.sqlQuery("select * from default_catalog.default_database.output"));

  • 注意1.11版本的执行sql的方法发生了改变,通过执行环境的executeSql(),executeInsert()等来进行插入或者执行sql语句

以上是“Flink SQL如何连接Hive并写入/读取数据”这篇文章的所有内容,感谢各位的阅读!希望分享的内容对大家有帮助,更多相关知识,欢迎关注创新互联行业资讯频道!


文章题目:FlinkSQL如何连接Hive并写入/读取数据
文章来源:http://bjjierui.cn/article/iecpdg.html

其他资讯