本文记录了学习使用Spark Streaming通过JDBC操作数据库的过程，源数据从Kafka中读取。

Kafka从0.10版本提供了一种新的消费者API，和0.8不同，因此Spark Streaming也提供了两种API与之对应，其中spark-streaming-kafka-0-8支持Kafka 0.8.2.1以后的Broker；spark-streaming-kafka-0-10支持0.10.0以上Broker，处于实验阶段。两者的对比如下表所示。

Spark Streaming集成Kafka的说明可以参考如下资料：

spark-streaming-kafka-0-10 http://spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html

spark-streaming-kafka-0-8 http://spark.apache.org/docs/latest/streaming-kafka-0-8-integration.html

1.2 运行组件

本文所写的例子基于JDK1.8和Scala 2.11，运行依赖组件的情况如下表所示。

组件	部署方式	IP地址	操作系统
spark-2.0.1	伪分布式	192.168.1.91	CentOS 7.1
Kafka-0.10.0.1	伪分布式	192.168.1.90	CentOS 7.1
postgresql-9.4.5	单机	192.168.1.213	CentOS 7.1

数据库中创建了一张名为kafka_message的表，共有三个字段，都是varchar类型。

CREATE TABLE kafka_message (

    timeseq varchar(16),

    thread varchar(32),

    message varchar(255)

);

2. 代码

2.1 pom.xml

依赖的lib如下。

<dependencies>

    <dependency>

        <groupId>org.apache.spark</groupId>

        <artifactId>spark-streaming_2.11</artifactId>

        <version>2.0.1</version>

    </dependency>

    <dependency>

        <groupId>org.apache.spark</groupId>

        <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>

        <version>2.0.1</version>

    </dependency>

    <dependency>

        <groupId>com.jolbox</groupId>

        <artifactId>bonecp</artifactId>

        <version>0.8.0.RELEASE</version>

    </dependency>

    <dependency>

        <groupId>postgresql</groupId>

        <artifactId>postgresql</artifactId>

        <version>9.1-901-1.jdbc4</version>

    </dependency>

</dependencies>

2.2 数据库连接池

import java.sql.Connection

import com.jolbox.bonecp.{BoneCP, BoneCPConfig}

import org.slf4j.LoggerFactory

 /**

  *  数据库连接池，使用了BoneCP

  */

object ConnectionPool {

  val logger = LoggerFactory.getLogger(this.getClass)

  //连接池配置

  private val connectionPool: Option[BoneCP] = {

    try{

      Class.forName("org.postgresql.Driver")

      val config = new BoneCPConfig()

      config.setJdbcUrl("jdbc:postgresql://192.168.1.213/yourdb")

      config.setUsername("postgres")

      config.setPassword("******")

      config.setLazyInit(true)

      config.setMinConnectionsPerPartition(3)

      config.setMaxConnectionsPerPartition(5)

      config.setPartitionCount(5)

      config.setCloseConnectionWatch(true)

      config.setLogStatementsEnabled(false)

      Some(new BoneCP(config))

    }catch {

      case exception: Exception =>

        logger.warn("Create Connection Error: \n" + exception.printStackTrace())

        None

    }

  }

  // 获取数据库连接

  def getConnection: Option[Connection] = {

    connectionPool match {

      case Some(pool) => Some(pool.getConnection)

      case None => None

    }

  }

  // 释放数据库连接

  def closeConnection(connection:Connection): Unit = {

    if(!connection.isClosed) {

      connection.close()

    }

  }

2.3 Kafka -> Spark-Streaming -> JDBC

Spark Streaming从Kafka中读取数据，并把数据写入数据库。SPark Streaming编程的基本顺序是：

创建Spark Streaming上下文
从数据源接口创建DStream
在DStream上做转换（Transformations）
指定计算结果存储的位置
启动计算

代码如下，详见注释。

import java.sql.Connection

import org.apache.kafka.common.serialization.StringDeserializer

import org.apache.spark.SparkConf

import org.apache.spark.streaming.kafka010.ConsumerStrategies._

import org.apache.spark.streaming.kafka010.KafkaUtils

import org.apache.spark.streaming.kafka010.LocationStrategies._

import org.apache.spark.streaming.{Seconds, StreamingContext}

import org.slf4j.LoggerFactory

/**

  * 从Kafka中读取数据，并把数据写入数据库。

  */

object KafkaToDB {

  val logger = LoggerFactory.getLogger(this.getClass)

  def main(args: Array[String]): Unit = {

    // 参数校验

    if (args.length < 2) {

      System.err.println(

        s"""

           |Usage: KafkaToDB <brokers> <topics>

           |  <brokers> broker列表，至少1个，以英文逗号分割

           |  <topics> topic列表，至少一个，以英文逗号分割

           |""".stripMargin)

      System.exit(1)

    }

    // 处理参数

    val Array(brokers, topics) = args

    val topicSet: Set[String] = topics.split(",").toSet

    val kafkaParams: Map[String, Object] = Map[String, Object](

      "bootstrap.servers" -> brokers,

      "key.deserializer" -> classOf[StringDeserializer],

      "value.deserializer" -> classOf[StringDeserializer],

      "group.id" -> "example",

      "auto.offset.reset" -> "latest",

      "enable.auto.commit" -> (false: java.lang.Boolean)

    )

    // 1.创建上下文，以每1秒间隔的数据作为一个批次

    val sparkConf = new SparkConf().setAppName("KafkaToDB")

    val streamingContext = new StreamingContext(sparkConf, Seconds(1))

    // 2.创建输入流，获取数据。流操作基于DStream，InputDStream继承于DStream

    val stream = KafkaUtils.createDirectStream[String, String](

      streamingContext,

      PreferConsistent,

      Subscribe[String, String](topicSet, kafkaParams)

    )

    // 3. DStream上的转换操作

    // 取消息中的value数据，以英文逗号分割，并转成Tuple3

    val values = stream.map(_.value.split(","))

      .filter(x => x.length == 3)

      .map(x => new Tuple3[String, String, String](x(0), x(1), x(2)))

    // 输入前10条到控制台，方便调试

    values.print()

    // 4.同foreachRDD保存到数据库

    val sql = "insert into kafka_message(timeseq, thread, message) values (?,?,?)"

    values.foreachRDD(rdd => {

      val count = rdd.count()

      println("-----------------count:" + count)

      if (count > 0) {

        rdd.foreachPartition(partitionOfRecords => {

          val conn = ConnectionPool.getConnection.orNull

          if (conn != null) {

            partitionOfRecords.foreach(data => insert(conn, sql, data))

            ConnectionPool.closeConnection(conn)

          }

        })

      }

    })

    // 5. 启动计算

    streamingContext.start()

    streamingContext.awaitTermination() // 等待中断结束计算

  }

  /**

    * 保存数据到数据库

    *

    * @param conn 数据库连接

    * @param sql  prepared statement sql

    * @param data 要保存的数据，Tuple3结构

    */

  def insert(conn: Connection, sql: String, data: (String, String, String)): Unit = {

    try {

      val ps = conn.prepareStatement(sql)

      ps.setString(1, data._1)

      ps.setString(2, data._2)

      ps.setString(3, data._3)

      ps.executeUpdate()

      ps.close()

    } catch {

      case e: Exception =>

        logger.error("Error in execution of insert. " + e.getMessage)

    }

  }

}

3. 任务运行

3.1 数据库驱动配置

由于本次程序运行环境是spark standalone 的伪分布式，指定SPARK_CLASSPATH时，相当于同时指定了driver和executor的classpath。

编辑spark-env.sh。

vi $SPARK_HOME/conf/spark-env.sh

输入以下内容，注意把postgresql驱动包放在对应的位置。

export SPARK_CLASSPATH=$SPARK_CLASSPATH:/opt/jobs/postgresql-9.1-901-1.jdbc4.jar

实际上，Spark已经不推荐使用SPARK_CLASSPATH参数了，启动的时候，我们会发现如下的日志，提示我们用spark.executor.extraClassPath和spark.driver.extraClassPath来代替。如果是spark local模式，只需指定spark.driver.extraClassPath即可。

日志如下。

16/10/21 15:15:33 WARN SparkConf:

SPARK_CLASSPATH was detected (set to ':/opt/jobs/postgresql-9.1-901-1.jdbc4.jar').

This is deprecated in Spark 1.0+.

Please instead use:

 - ./spark-submit with --driver-class-path to augment the driver classpath

 - spark.executor.extraClassPath to augment the executor classpath

16/10/21 15:15:33 WARN SparkConf: Setting 'spark.executor.extraClassPath' to ':/opt/jobs/

postgresql-9.1-901-1.jdbc4.jar' as a work-around.

16/10/21 15:15:33 WARN SparkConf: Setting 'spark.driver.extraClassPath' to ':/opt/jobs/

postgresql-9.1-901-1.jdbc4.jar' as a work-around.

3.2 启动任务

$SPARK_HOME/bin/spark-submit \

--master spark://192.168.1.91:7077

--class com.xxx.streaming.KafkaToDB \

 spark-streaming-demo.jar 192.168.1.90:9092 my-topic

在192.168.1.90上，通过kafka命令发送消息，命令如下：

bin/kafka-console-producer.sh --broker-list 192.168.1.90:9092 --topic mytopic

消息如下：

1475589918658    thread-15    msg-0

1475589920177    thread-15    msg-1

之后，我们可以在控制台看到消息被输出，也可以在数据库中查询到这些数据。

4.Spark程序找不到JDBC驱动的问题

控制台曾经报出找不到JDBC驱动的异常，日志如下。

java.sql.SQLException: No suitable driver found for jdbc:……

该问题可以用前文所述的数据库驱动配置办法解决。在这里推荐一篇博客，对这个问题给出的很好的解释。
http://www.codexiu.cn/spark/blog/12672/

博客来源：https://www.jianshu.com/p/a73c0c95d2fe

秒客网

Spark Streaming通过JDBC操作数据库