Я хочу сделать стриминг в Apache Flink для Kafka -> Flink -> HIVE

Question

Я хочу выполнить потоковую передачу в Apache Flink для выполнения Kafka -> Flink -> HIVE в Apache Flink (Scala). Кто-нибудь может дать образец кода, поскольку их официальный документ не очень ясен для понимания.

Это должен быть потоковый процесс.

Answer 1

Чтобы получить помощь по началу работы с Table API, Создание отчетов в реальном времени с Table API - это руководство, которому вы можете следовать. Он в Java, но API Scala не сильно отличается.

Этот является примером использования SQL для чтения из Kafka и записи в Hive. Чтобы сделать то же самое с Scala, вы можете обернуть операторы SQL с tableEnv.executeSql(...), как в

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

или

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

Если вам нужно выполнить несколько вставок, то вам нужно будет сделать это немного по-другому, используя StatementSet. Подробности см. В документации по ссылкам ниже.

См. Выполнить оператор CREATE , Запустить оператор INSERT , Apache Kafka SQL Connector и Запись в Hive .

Если вы застряли, покажите нам, что вы пробовали и как это не удается.