我想在 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。

如果您遇到困难，请向我们展示您的尝试以及失败的原因。