我想在Apache Flink中做流式作业来做Kafka -> Flink -> HIVE

Question

我想在Apache Flink中做流式作业，在Apache Flink(Scala)中做Kafka -> Flink -> HIVE。谁能给代码样本，因为他们的官方文件不是很清楚。

这应该是流过程。

Answer 1

要获得有关Table API入门的帮助，可以参考Real Time Reporting with the Table API教程。它是用Java编写的，但是Scala API并没有太大的不同。

This是一个使用SQL从Kafka读取和写入Hive的示例。要在Scala中执行相同的操作，可以使用tableEnv.executeSql(...)包装SQL语句，如

tableEnv.executeSql("CREATE TABLE Orders (`user` BIGINT, product STRING, amount INT) WITH (...)")

或

val tableResult1 = tEnv.executeSql("INSERT INTO ...")

如果您需要执行多个插入，那么您需要使用StatementSet来做一些不同的操作。有关详细信息，请参阅下面链接的文档。

请参见Run a CREATE statement、Run an INSERT statement、Apache Kafka SQL Connector和Writing to Hive。

如果你被卡住了，告诉我们你尝试过什么，以及它是如何失败的。