这篇文章主要介绍Flink SQL如何连接Hive并写入/读取数据,文中介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们一定要看完!
<properties>
<flink.version>1.11.2</flink.version>
<scala.version>2.11</scala.version>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-scala_${scala.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka-0.11_${scala.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_${scala.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-api-java-bridge_${scala.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-planner-blink_${scala.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<!-- 添加flink table api 集成Hive的依赖-->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-hive_${scala.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<!-- Hive Dependency -->
<dependency>
<groupId>org.apache.hive</groupId>
<artifactId>hive-exec</artifactId>
<version>2.1.1</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-shaded-hadoop-2-uber</artifactId>
<version>2.6.5-7.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-json</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-elasticsearch7_${scala.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-csv</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>2.4.0</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-annotations</artifactId>
<version>2.4.0</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-core</artifactId>
<version>2.4.0</version>
</dependency>
</dependencies>
EnvironmentSettings bbSettings = EnvironmentSettings.newInstance()
.useBlinkPlanner()
.inBatchMode()
.build();
TableEnvironment bbTableEnv = TableEnvironment.create(bbSettings);
经过实际测试,目前HiveTableSink 不支持流式写入(未实现 AppendStreamTableSink),必须是批处理环境才可以往hive里面写入数据,而不能将流式数据写入hive。例如将kafka创建一张临时表,然后将表中的数据流持续插入hive,这是不可以的,官网上1.11版本通过flink sql-client可以实现hive的流式写入,还有待验证。
// 连接外部文件
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并写入/读取数据”这篇文章的所有内容,感谢各位的阅读!希望分享的内容对大家有帮助,更多相关知识,欢迎关注亿速云行业资讯频道!
亿速云「云服务器」,即开即用、新一代英特尔至强铂金CPU、三副本存储NVMe SSD云盘,价格低至29元/月。点击查看>>
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。
原文链接:https://my.oschina.net/u/4252423/blog/4653452