Spark支持四种方式从数据库中读取数据

骑猪看日落 2022-07-16 04:42 360阅读 0赞

目前 [Spark][]支持四种方式从数据库中读取数据，这里以Mysql为例进行介绍。

文章目录 \[[hide][]\]

*  [1 一、不指定查询条件][1]
 *  [2 二、指定数据库字段的范围][2]
 *  [3 三、根据任意字段进行分区][3]
 *  [4 四、通过load获取][4 _load]

## 一、不指定查询条件 ##

这个方式链接MySql的函数原型是：

<table> 
 <tbody> 
  <tr> 
   <td> 
    <div> 
     <div> 
      <code>def</code> 
      <code> jdbc(url</code> 
      <code>:</code> 
      <code>String, table</code> 
      <code>:</code> 
      <code>String, properties</code> 
      <code>:</code> 
      <code>Properties)</code> 
      <code>:</code> 
      <code> DataFrame</code> 
     </div> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

我们只需要提供Driver的url，需要查询的表名，以及连接表相关属性properties。下面是具体例子：

<table> 
 <tbody> 
  <tr> 
   <td> 
    <div> 
     <div> 
      <code>val</code> 
      <code> url </code> 
      <code>=</code> 
      <code>"jdbc:<a rel="nofollow">mysql://www.iteblog.com:3306/iteblog?user=iteblog&amp;password=iteblog</a>"</code> 
     </div> 
     <div>
       &nbsp; 
     </div> 
     <div> 
      <code>val</code> 
      <code> prop </code> 
      <code>=</code> 
      <code>new</code> 
      <code>Properties()</code> 
     </div> 
     <div> 
      <code>val</code> 
      <code> df </code> 
      <code>=</code> 
      <code>sqlContext.read.jdbc(url, </code> 
      <code>"iteblog"</code> 
      <code>, prop )</code> 
     </div> 
     <div>
       &nbsp; 
     </div> 
     <div> 
      <code>println(df.count())</code> 
     </div> 
     <div> 
      <code>println(df.rdd.partitions.size)</code> 
     </div> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

我们运行上面的程序，可以看到`df.rdd.partitions.size`输出结果是1，这个结果的含义是iteblog表的所有数据都是由RDD的一个分区处理的，所以说，如果你这个表很大，很可能会出现OOM

<table> 
 <tbody> 
  <tr> 
   <td> 
    <div> 
     <div> 
      <code>WARN TaskSetManager</code> 
      <code>:</code> 
      <code>Lost task </code> 
      <code>0.0</code> 
      <code> in stage </code> 
      <code>1.0</code> 
      <code>(TID </code> 
      <code>14</code> 
      <code>, spark</code> 
      <code>047219</code> 
      <code>)</code> 
      <code>:</code> 
     </div> 
     <div> 
      <code>&nbsp;</code> 
      <code>java.lang.OutOfMemoryError</code> 
      <code>:</code> 
      <code>GC overhead limit exceeded at com.mysql.jdbc.MysqlIO.reuseAndReadPacket(MysqlIO.java</code> 
      <code>:</code> 
      <code>3380</code> 
      <code>)</code> 
     </div> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

这种方式在数据量大的时候不建议使用。

[![Spark_Mysql.png][]][Spark_Mysql.png]  
如果想及时了解 [Spark][]、Hadoop或者Hbase相关的文章，欢迎关注微信公共帐号： **iteblog\_hadoop**

## 二、指定数据库字段的范围 ##

这种方式就是通过指定数据库中某个字段的范围，但是遗憾的是，这个字段必须是数字，来看看这个函数的函数原型：

<table> 
 <tbody> 
  <tr> 
   <td> 
    <div> 
     <div> 
      <code>def</code> 
      <code> jdbc(</code> 
     </div> 
     <div> 
      <code>&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code>url</code> 
      <code>:</code> 
      <code>String,</code> 
     </div> 
     <div> 
      <code>&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code>table</code> 
      <code>:</code> 
      <code>String,</code> 
     </div> 
     <div> 
      <code>&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code>columnName</code> 
      <code>:</code> 
      <code>String,</code> 
     </div> 
     <div> 
      <code>&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code>lowerBound</code> 
      <code>:</code> 
      <code>Long,</code> 
     </div> 
     <div> 
      <code>&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code>upperBound</code> 
      <code>:</code> 
      <code>Long,</code> 
     </div> 
     <div> 
      <code>&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code>numPartitions</code> 
      <code>:</code> 
      <code>Int,</code> 
     </div> 
     <div> 
      <code>&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code>connectionProperties</code> 
      <code>:</code> 
      <code>Properties)</code> 
      <code>:</code> 
      <code> DataFrame</code> 
     </div> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

前两个字段的含义和方法一类似。columnName就是需要分区的字段，这个字段在数据库中的类型必须是数字；lowerBound就是分区的下界；upperBound就是分区的上界；numPartitions是分区的个数。同样，我们也来看看如何使用：

<table> 
 <tbody> 
  <tr> 
   <td> 
    <div> 
     <div> 
      <code>val</code> 
      <code> lowerBound </code> 
      <code>=</code> 
      <code> 1</code> 
     </div> 
     <div> 
      <code>val</code> 
      <code> upperBound </code> 
      <code>=</code> 
      <code> 100000</code> 
     </div> 
     <div> 
      <code>val</code> 
      <code> numPartitions </code> 
      <code>=</code> 
      <code> 5</code> 
     </div> 
     <div> 
      <code>val</code> 
      <code> url </code> 
      <code>=</code> 
      <code>"jdbc:<a rel="nofollow">mysql://www.iteblog.com:3306/iteblog?user=iteblog&amp;password=iteblog</a>"</code> 
     </div> 
     <div>
       &nbsp; 
     </div> 
     <div> 
      <code>val</code> 
      <code> prop </code> 
      <code>=</code> 
      <code>new</code> 
      <code>Properties()</code> 
     </div> 
     <div> 
      <code>val</code> 
      <code> df </code> 
      <code>=</code> 
      <code>sqlContext.read.jdbc(url, </code> 
      <code>"iteblog"</code> 
      <code>, </code> 
      <code>"id"</code> 
      <code>, lowerBound, upperBound, numPartitions, prop)</code> 
     </div> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

这个方法可以将iteblog表的数据分布到RDD的几个分区中，分区的数量由numPartitions参数决定，在理想情况下，每个分区处理相同数量的数据，我们在使用的时候不建议将这个值设置的比较大，因为这可能导致数据库挂掉！但是根据前面介绍，这个函数的缺点就是只能使用整形数据字段作为分区关键字。

这个函数在极端情况下，也就是设置将numPartitions设置为1，其含义和第一种方式一致。

## 三、根据任意字段进行分区 ##

基于前面两种方法的限制，[Spark][]还提供了根据任意字段进行分区的方法，函数原型如下：

这个函数相比第一种方式多了predicates参数，我们可以通过这个参数设置分区的依据，来看看例子：

<table> 
 <tbody> 
  <tr> 
   <td> 
    <div> 
     <div> 
      <code>val</code> 
      <code> predicates </code> 
      <code>=</code> 
      <code> Array[String](</code> 
      <code>"reportDate &lt;= '2014-12-31'"</code> 
      <code>, </code> 
     </div> 
     <div> 
      <code>&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code>"reportDate &gt; '2014-12-31' and reportDate &lt;= '2015-12-31'"</code> 
      <code>)</code> 
     </div> 
     <div> 
      <code>val</code> 
      <code> url </code> 
      <code>=</code> 
      <code>"jdbc:<a rel="nofollow">mysql://www.iteblog.com:3306/iteblog?user=iteblog&amp;password=iteblog</a>"</code> 
     </div> 
     <div>
       &nbsp; 
     </div> 
     <div> 
      <code>val</code> 
      <code> prop </code> 
      <code>=</code> 
      <code>new</code> 
      <code>Properties()</code> 
     </div> 
     <div> 
      <code>val</code> 
      <code> df </code> 
      <code>=</code> 
      <code>sqlContext.read.jdbc(url, </code> 
      <code>"iteblog"</code> 
      <code>, predicates, prop)</code> 
     </div> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

最后rdd的分区数量就等于predicates.length。

## 四、通过load获取 ##

Spark还提供通过load的方式来读取数据。

<table> 
 <tbody> 
  <tr> 
   <td> 
    <div> 
     <div> 
      <code>sqlContext.read.format(</code> 
      <code>"jdbc"</code> 
      <code>).options(</code> 
     </div> 
     <div> 
      <code>&nbsp;&nbsp;</code> 
      <code>Map(</code> 
      <code>"url"</code> 
      <code>-&gt; </code> 
      <code>"jdbc:<a rel="nofollow">mysql://www.iteblog.com:3306/iteblog?user=iteblog&amp;password=iteblog</a>"</code> 
      <code>,</code> 
     </div> 
     <div> 
      <code>&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code>"dbtable"</code> 
      <code>-&gt; </code> 
      <code>"iteblog"</code> 
      <code>)).load()</code> 
     </div> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

options函数支持url、driver、dbtable、partitionColumn、lowerBound、upperBound以及numPartitions选项，细心的同学肯定发现这个和方法二的参数一致。是的，其内部实现原理部分和方法二大体一致。同时load方法还支持json、orc等数据源的读取。

[Spark]: https://www.iteblog.com/archives/tag/spark
[hide]: https://www.iteblog.com/archives/1560#
[1]: https://www.iteblog.com/archives/1560#i
[2]: https://www.iteblog.com/archives/1560#i-2
[3]: https://www.iteblog.com/archives/1560#i-3
[4 _load]: https://www.iteblog.com/archives/1560#load
[Spark_Mysql.png]: /images/20220716/ef18e4fc049f4e069cbd60d2e8183f04.png