scala 消费topic写入hive

Scala 语言提供了一种灵活的方式来消费 Kafka Topic 并将数据写入 Hive 数据仓库。

首先，我们需要使用 Kafka 的 Scala 客户端库来消费 Topic 中的数据。可以使用如下代码示例创建一个消费者来读取 Kafka Topic 中的消息：

import java.util.Properties
import org.apache.kafka.clients.consumer.KafkaConsumer

val props = new Properties()
props.put("bootstrap.servers", "localhost:9092")
props.put("group.id", "my-consumer-group")
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")

val consumer = new KafkaConsumer[String, String](props)
consumer.subscribe(Collections.singletonList("my-topic"))

while (true) {
  val records = consumer.poll(Duration.ofMillis(100))
  for (record <- records.asScala) {
    val message = record.value()
    // 处理消息，写入 Hive
    // ...
  }
}

上述代码中，我们创建了一个 KafkaConsumer 对象，配置了 Kafka 服务的地址、消费者组、序列化器等信息。然后使用 `subscribe` 方法订阅了一个 Topic，接着使用 `poll` 方法从 Topic 中拉取消息。在消息的处理过程中，可以将数据写入 Hive 数据仓库。

对于将数据写入 Hive，我们可以使用 Hive JDBC 驱动来连接 Hive，并执行 SQL 插入语句。以下是一个使用 Hive JDBC 连接并插入数据的示例：

import java.sql.{Connection, DriverManager, Statement}

// JDBC 连接 Hive
val driverName = "org.apache.hive.jdbc.HiveDriver"
Class.forName(driverName)
val connection = DriverManager.getConnection("jdbc:hive2://localhost:10000/default", "", "")
val statement = connection.createStatement()

// 插入数据到 Hive 表
val sql = "INSERT INTO my_table VALUES ('data')"
statement.execute(sql)

// 关闭连接
statement.close()
connection.close()

在上述示例代码中，我们使用 JDBC 驱动程序连接到 Hive，并创建了一个 Statement 对象来执行 SQL 语句。然后，我们可以使用 `execute` 方法来执行插入语句将数据写入 Hive 表中。

综上所述，使用 Scala 可以方便地消费 Kafka Topic，并将数据写入 Hive 数据仓库。只需使用 Kafka 的 Scala 客户端库和 Hive JDBC 驱动，分别处理消费和写入操作即可。