sparkStreaming 连接数据库遇到的问题和解决方法

问题导读：
1. 在使用destream.foreachrdd 连接数据库时经常出现的错误时什么？
2. 如何解决上面的错误？




解决方案：


使用destream.foreachrdd经常出现的两个错误

当我们遇到需要用sparkStreaming来保存处理的数据时，自然会想到用ojdbc.但是当我们真正去实现她的时候会发现，用Spark Streaming来处理数据库连接是一件有点麻烦的事情，下面是针对spark-streaming_2.11来处理的。
destream.foreachrdd是一个强大的可以将数据发送到外部系统的方法。然而，在使用她的时候我们常常会犯一些无法理解的错误。


错误一：

[mw_shl_code=scala,true]dstream.foreachRDD { rdd =>
  val connection = createNewConnection()  // executed at the driver
  rdd.foreach { record =>
    connection.send(record) // executed at the worker
  }
}[/mw_shl_code]

这种写法是不正确的，如果这样写，连接对象会被序列化到worker里，这样就不能连接数据库了。

错误二：

[mw_shl_code=scala,true]dstream.foreachRDD { rdd =>
  rdd.foreach { record =>
    val connection = createNewConnection()
    connection.send(record)
    connection.close()
  }
}[/mw_shl_code]

这种情况会导致另一种问题。会为每一个记录都创建一个新的连接，如果一次处理的数据过多就会创建很多数据库连接，这样就造成了服务器压力过大，资源消耗严重，容易使集群崩溃。

正确连接方法

正确连接方式：

[mw_shl_code=scala,true]dstream.foreachRDD { rdd =>
  rdd.foreachPartition { partitionOfRecords =>
    val connection = createNewConnection()
    partitionOfRecords.foreach(record => connection.send(record))
    connection.close()
  }
}
[/mw_shl_code]


这种方式是为每一个partition分区分配一个连接，不同的partition拥有不同的连接。

优化版：


[mw_shl_code=scala,true]dstream.foreachRDD { rdd =>
  rdd.foreachPartition { partitionOfRecords =>
    // ConnectionPool is a static, lazily initialized pool of connections
    val connection = ConnectionPool.getConnection()
    partitionOfRecords.foreach(record => connection.send(record))
    ConnectionPool.returnConnection(connection)  // return to the pool for future reuse
  }
}[/mw_shl_code]

这种需要创建一个数据库连接池，每次需要使用数据库连接的时候都会先从数据库连接池中获取数据库连接。这是一种lazily的模式。节省程序创建数据库连接的开销。

.........关于spark streaming 向外部导入数据的资料，全是直接翻译官网的。没意思

学习了