使用Spark读写CSV格式文件

爱被打了一巴掌 2022-07-21 00:23 296阅读 0赞

转载地址：[http://blog.csdn.net/zy\_zhengyang/article/details/49382715][http_blog.csdn.net_zy_zhengyang_article_details_49382715]

CSV格式的文件也称为逗号分隔值（Comma-Separated Values，CSV，有时也称为字符分隔值，因为分隔字符也可以不是逗号。在本文中的CSV格式的数据就不是简单的逗号分割的），其文件以纯文本形式存表格数据（数字和文本）。CSV文件由任意数目的记录组成，记录间以某种换行符分隔；每条记录由字段组成，字段间的分隔符是其它字符或字符串，最常见的是逗号或制表符。通常，所有记录都有完全相同的字段序列。

本篇文章将介绍如何使用[Spark][] 1.3+的外部数据源接口来自定义CSV输入格式的文件解析器。这个外部数据源接口是由databricks公司开发并开源的（地址：https://github.com/databricks/spark-csv），通过这个类库我们可以在[Spark][] SQL中解析并查询CSV中的数据。因为用到了[Spark][]的外部数据源接口，所以我们需要在Spark 1.3+上面使用。在使用之前，我们需要引入以下的依赖：

<table> 
 <tbody> 
  <tr> 
   <td><code>1</code></td> 
   <td><code>&lt;</code><code>dependency</code><code>&gt;</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>2</code></td> 
   <td><code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>&lt;</code><code>groupId</code><code>&gt;com.databricks&lt;/</code><code>groupId</code><code>&gt;</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>3</code></td> 
   <td><code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>&lt;</code><code>artifactId</code><code>&gt;spark-csv_2.10&lt;/</code><code>artifactId</code><code>&gt;</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>4</code></td> 
   <td><code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>&lt;</code><code>version</code><code>&gt;1.0.3&lt;/</code><code>version</code><code>&gt;</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>5</code></td> 
   <td><code>&lt;/</code><code>dependency</code><code>&gt;</code></td> 
  </tr> 
 </tbody> 
</table>

目前spark-csv\_2.10的最新版就是1.0.3。如果我们想在Spark shell里面使用，我们可以在`--jars`选项里面加入这个依赖，如下：

<table> 
 <tbody> 
  <tr> 
   <td><code>1</code></td> 
   <td><code>[iteblog@spark $] bin/spark-shell --packages com.databricks:spark-csv_2.10:1.0.3</code></td> 
  </tr> 
 </tbody> 
</table>

和[《Spark SQL整合PostgreSQL》][Spark SQL_PostgreSQL]文章中用到的load函数类似，在使用CSV类库的时候，我们需要在`options`中传入以下几个选项：

1、`path`：看名字就知道，这个就是我们需要解析的CSV文件的路径，路径支持通配符；  
　　2、`header`：默认值是false。我们知道，CSV文件第一行一般是解释各个列的含义的名称，如果我们不需要加载这一行，我们可以将这个选项设置为true；  
　　3、`delimiter`：默认情况下，CSV是使用英文逗号分隔的，如果不是这个分隔，我们就可以设置这个选项。  
　　4、`quote`：默认情况下的引号是'"'，我们可以通过设置这个选项来支持别的引号。  
　　5、`mode`：解析的模式。默认值是`PERMISSIVE`，支持的选项有  
　　　　（1）、`PERMISSIVE`：尝试解析所有的行，nulls are inserted for missing tokens and extra tokens are ignored.  
　　　　（2）、`DROPMALFORMED`：drops lines which have fewer or more tokens than expected  
　　　　（3）、`FAILFAST`: aborts with a RuntimeException if encounters any malformed line

## 如何使用 ##

### 1、在Spark SQL中使用 ###

我们可以通过注册临时表，然后使用纯SQL方式去查询CSV文件：

<table> 
 <tbody> 
  <tr> 
   <td><code>1</code></td> 
   <td><code>CREATE</code>&nbsp;<code>TABLE</code>&nbsp;<code>cars</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>2</code></td> 
   <td><code>USING com.databricks.spark.csv</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>3</code></td> 
   <td><code>OPTIONS (path&nbsp;</code><code>"cars.csv"</code><code>, header&nbsp;</code><code>"true"</code><code>)</code></td> 
  </tr> 
 </tbody> 
</table>

我们还可以在DDL中指定列的名字和类型，如下：

<table> 
 <tbody> 
  <tr> 
   <td><code>1</code></td> 
   <td><code>CREATE</code>&nbsp;<code>TABLE</code>&nbsp;<code>cars (yearMade&nbsp;</code><code>double</code><code>, carMake string, carModel string, comments string, blank string)</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>2</code></td> 
   <td><code>USING com.databricks.spark.csv</code></td> 
  </tr> 
 </tbody> 
</table>

### 2、通过[Scala][]方式 ###

推荐的方式是通过调用`SQLContext`的`load/save`函数来加载CSV数据：

<table> 
 <tbody> 
  <tr> 
   <td><code>1</code></td> 
   <td><code>import</code>&nbsp;<code>org.apache.spark.sql.SQLContext</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>3</code></td> 
   <td><code>val</code>&nbsp;<code>sqlContext&nbsp;</code><code>=</code>&nbsp;<code>new</code>&nbsp;<code>SQLContext(sc)</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>4</code></td> 
   <td><code>val</code>&nbsp;<code>df&nbsp;</code><code>=</code>&nbsp;<code>sqlContext.load(</code><code>"com.databricks.spark.csv"</code><code>, Map(</code><code>"path"</code>&nbsp;<code>-&gt;&nbsp;</code><code>"cars.csv"</code><code>,&nbsp;</code><code>"header"</code>&nbsp;<code>-&gt;&nbsp;</code><code>"true"</code><code>))</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>5</code></td> 
   <td><code>df.select(</code><code>"year"</code><code>,&nbsp;</code><code>"model"</code><code>).save(</code><code>"newcars.csv"</code><code>,&nbsp;</code><code>"com.databricks.spark.csv"</code><code>)</code></td> 
  </tr> 
 </tbody> 
</table>

当然，我们还可以使用`com.databricks.spark.csv._`的隐式转换：

<table> 
 <tbody> 
  <tr> 
   <td><code>1</code></td> 
   <td><code>import</code>&nbsp;<code>org.apache.spark.sql.SQLContext</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>2</code></td> 
   <td><code>import</code>&nbsp;<code>com.databricks.spark.csv.</code><code>_</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>4</code></td> 
   <td><code>val</code>&nbsp;<code>sqlContext&nbsp;</code><code>=</code>&nbsp;<code>new</code>&nbsp;<code>SQLContext(sc)</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>6</code></td> 
   <td><code>val</code>&nbsp;<code>cars&nbsp;</code><code>=</code>&nbsp;<code>sqlContext.csvFile(</code><code>"cars.csv"</code><code>)</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>7</code></td> 
   <td><code>cars.select(</code><code>"year"</code><code>,&nbsp;</code><code>"model"</code><code>).saveAsCsvFile(</code><code>"newcars.tsv"</code><code>)</code></td> 
  </tr> 
 </tbody> 
</table>

### 3、在Java中使用 ###

和在[Scala][]中使用类似，我们也推荐调用`SQLContext`类中 `load/save`函数

<table> 
 <tbody> 
  <tr> 
   <td><code>05</code></td> 
   <td><code>&nbsp;</code><code>* bolg:&nbsp;<a href="http://www.iteblog.com/" style="color:rgb(106,57,6); text-decoration:none" rel="nofollow">http://www.iteblog.com</a></code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>06</code></td> 
   <td><code>&nbsp;</code><code>* 本文地址：<a href="http://www.iteblog.com/archives/1380" style="color:rgb(106,57,6); text-decoration:none" rel="nofollow">http://www.iteblog.com/archives/1380</a></code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>07</code></td> 
   <td><code>&nbsp;</code><code>* 过往记忆博客，专注于hadoop、hive、spark、shark、flume的技术博客，大量的干货</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>08</code></td> 
   <td><code>&nbsp;</code><code>* 过往记忆博客微信公共帐号：iteblog_hadoop</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>11</code></td> 
   <td><code>import</code>&nbsp;<code>org.apache.spark.sql.SQLContext</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>13</code></td> 
   <td><code>SQLContext sqlContext =&nbsp;</code><code>new</code>&nbsp;<code>SQLContext(sc);</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>15</code></td> 
   <td><code>HashMap&lt;String, String&gt; options =&nbsp;</code><code>new</code>&nbsp;<code>HashMap&lt;String, String&gt;();</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>16</code></td> 
   <td><code>options.put(</code><code>"header"</code><code>,&nbsp;</code><code>"true"</code><code>);</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>17</code></td> 
   <td><code>options.put(</code><code>"path"</code><code>,&nbsp;</code><code>"cars.csv"</code><code>);</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>19</code></td> 
   <td><code>DataFrame df = sqlContext.load(</code><code>"com.databricks.spark.csv"</code><code>, options);</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>20</code></td> 
   <td><code>df.select(</code><code>"year"</code><code>,&nbsp;</code><code>"model"</code><code>).save(</code><code>"newcars.csv"</code><code>,&nbsp;</code><code>"com.databricks.spark.csv"</code><code>);</code></td> 
  </tr> 
 </tbody> 
</table>

在Java或者是Scala中，我们可以通过CsvParser里面的函数来读取CSV文件:

<table> 
 <tbody> 
  <tr> 
   <td><code>1</code></td> 
   <td><code>import</code>&nbsp;<code>com.databricks.spark.csv.CsvParser;</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>2</code></td> 
   <td><code>SQLContext sqlContext =&nbsp;</code><code>new</code>&nbsp;<code>org.apache.spark.sql.SQLContext(sc);</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>4</code></td> 
   <td><code>DataFrame cars = (</code><code>new</code>&nbsp;<code>CsvParser()).withUseHeader(</code><code>true</code><code>).csvFile(sqlContext,&nbsp;</code><code>"cars.csv"</code><code>);</code></td> 
  </tr> 
 </tbody> 
</table>

### 4、在[Python][]中使用 ###

在[Python][]中，我们也可以使用`SQLContext`类中 `load/save`函数来读取和保存CSV文件：

<table> 
 <tbody> 
  <tr> 
   <td><code>1</code></td> 
   <td><code>from</code>&nbsp;<code>pyspark.sql&nbsp;</code><code>import</code>&nbsp;<code>SQLContext</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>2</code></td> 
   <td><code>sqlContext&nbsp;</code><code>=</code>&nbsp;<code>SQLContext(sc)</code></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><code>4</code></td> 
   <td><code>df&nbsp;</code><code>=</code>&nbsp;<code>sqlContext.load(source</code><code>=</code><code>"com.databricks.spark.csv"</code><code>, header</code><code>=</code><code>"true"</code><code>, path&nbsp;</code><code>=</code>&nbsp;<code>"cars.csv"</code><code>)</code></td> 
  </tr> 
 </tbody> 
</table>

[http_blog.csdn.net_zy_zhengyang_article_details_49382715]: http://blog.csdn.net/zy_zhengyang/article/details/49382715
[Spark]: http://www.iteblog.com/archives/tag/spark
[Spark SQL_PostgreSQL]: http://www.iteblog.com/archives/1369
[Scala]: http://www.iteblog.com/archives/tag/scala
[Python]: http://www.iteblog.com/archives/tag/python