Scala读取卡夫卡主题并写入MySQL表

基础概念

Scala: 是一种多范式的编程语言，设计初衷是要集成面向对象编程和函数式编程的各种优点。

Kafka: 是一个分布式流处理平台，用于构建实时数据管道和流应用程序。

MySQL: 是一个关系型数据库管理系统，广泛用于Web应用和内部系统中。

类型与应用场景

类型:

生产者: 负责将数据发送到Kafka。
消费者: 负责从Kafka读取数据。
数据库连接: 用于将数据写入MySQL。

应用场景:

实时数据处理和分析。
日志收集和监控系统。
用户行为跟踪和分析。

示例代码

以下是一个简单的Scala程序，展示如何使用akka-stream-kafka库从Kafka读取数据，并使用JDBC将数据写入MySQL。

import akka.actor.ActorSystem
import akka.kafka.scaladsl.Consumer
import akka.kafka.{ConsumerSettings, Subscriptions}
import akka.stream.scaladsl.Sink
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.common.serialization.StringDeserializer
import java.sql.{Connection, DriverManager, PreparedStatement}

object KafkaToMySQL {
  def main(args: Array[String]): Unit = {
    implicit val system = ActorSystem()

    val bootstrapServers = "localhost:9092"
    val topic = "test-topic"
    val groupId = "group1"

    val consumerSettings = ConsumerSettings(system, new StringDeserializer, new StringDeserializer)
      .withBootstrapServers(bootstrapServers)
      .withGroupId(groupId)
      .withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")

    Consumer.plainSource(consumerSettings, Subscriptions.topics(topic))
      .map(record => record.value())
      .runWith(Sink.foreach { value =>
        writeToMySQL(value)
      })
  }

  def writeToMySQL(data: String): Unit = {
    val url = "jdbc:mysql://localhost:3306/mydatabase"
    val username = "user"
    val password = "password"

    try (Connection conn = DriverManager.getConnection(url, username, password)) {
      val sql = "INSERT INTO mytable (data) VALUES (?)"
      val pstmt = conn.prepareStatement(sql)
      pstmt.setString(1, data)
      pstmt.executeUpdate()
    } catch {
      case e: Exception => e.printStackTrace()
    }
  }
}