Spark 是专为大规模数据分析处理而设计的开源分布式计算框架。本文介绍如何配置 EMR 中的 Spark 服务使用 CloudFS。

前提条件

• 开通大数据文件存储服务并创建文件存储实例，获取挂载点信息。具体操作，请参见开通大数据文件存储。
• 完成 E-MapReduce 中的集群创建。具体操作，请参见 E-MapReduce 集群创建。
• 准备一个测试文件。

步骤一：配置 CloudFS 服务

1. 连接 E-MapReduce 集群，连接方式如下：
   • 使用本地终端 ssh 连接集群节点管理 master 的公网 ip。
   • 使用同区域下的云服务器实例连接集群节点管理 master 的内网 ip。
2. 执行以下命令，下载1.5.3.49版本的CloudfsSDK包。

wget https://cloudfs.tos-cn-beijing.volces.com/sdk/prod/cloudfs-client-1.5.3.49.tar.gz

2. 执行以下命令，解压CloudfsSDK包。

tar -zxvf cloudfs-client-1.5.3.49.tar.gz

3. 将解压文件cloudfs-client/lib/cloudfs-sdk-1.5.3.49.jar拷贝至集群/share/hadoop/hdfs路径下。

   cp {YOUR_DOWNLOAD_PATH}/cloudfs-client/lib/cloudfs-sdk-1.5.3.49.jar /opt/{Directory}/hadoop/share/hadoop/hdfs/
   

4. 配置core-site.xml文件。

   1. 执行以下命令打开 Hadoop 安装目录下的core-site.xml文件：
      vim {hadoop_安装目录}/hadoop/conf/core-site.xml
   2. 在core-site.xml中添加配置：

<property>
  <name>fs.defaultFS</name>
  <value>cfs://xxxx.cfs-cn-beijing.ivolces.com</value>
<!-- 填写获取的挂载点地址 -->
</property>

<property>
  <name>fs.cfs.impl</name>
  <value>com.volcengine.cloudfs.fs.CfsFileSystem</value>
</property>

<property>
  <name>fs.AbstractFileSystem.cfs.impl</name>
  <value>com.volcengine.cloudfs.fs.CFS</value>
</property>  

<property>
  <name>cfs.access.key</name>
  <value>AKxxxxxxxxxxx</value>
<!-- 填写访问密钥ID -->
</property>
<property>
  <name>cfs.secret.key</name>
  <value>SKxxxxxxxxxxx</value>
<!-- 填写私有访问密钥-->
</property>

<!-- 可选：如果使用的是 STS Token，需要填写 -->
<property>
  <name>cfs.security.token</name>
  <value>STSTokenxxxxxxxx</value>
</property>

<!-- 可选：如果开启缓存加速，需要配置缓存加速接入的 VPC 的网段 -->
<property>
  <name>cfs.client.network.segment</name>
  <value><VPC 网段，例如 192.168.0.0/16></value>
</property>

步骤二：配置 Spark

5. 将解压后的 SDK 目录下的cloudfs-client/lib/cloudfs-sdk-1.5.3.49.jar文件复制到 Spark 的/{Directory}/spark/jars/目录下。

   cp {YOUR_DOWNLOAD_PATH}/cloudfs-client/lib/cloudfs-sdk-1.5.3.49.jar /{Directory}/spark/jars/
   

6. 重启 Spark 服务。

   1. 登录火山引擎 E-MapReduce 控制台。
   2. 在集群管理页面，找到目标 E-MapReduce 集群。
   3. 单击服务列表找到 Spark 服务。
   4. 单击列表页面右侧重启按钮，选择重启全部组件。重启后需重新连接集群。

步骤三：验证 Spark 配置

使用 Spark 处理 CloudFS 中的数据有两种方式。具体步骤如下：

• 使用Spark-sql处理 CloudFS 的数据。
  使Spark-sql处理 CloudFS 中的数据的方式和 Hivesql 类似，只需要建表的时候指定 locatition 至 CloudFS 文件系统即可。详细操作，可参见使用 Hive 处理 CloudFS 中的数据。

• 使用Spark-shell处理 CloudFS 的数据。具体操作步骤如下：

  1. 进入 Spark 的bin目录下，执行以下命令进去界面。

  ./spark-shell
  

  如果返回以下信息，则表示执行成功。
  
  2. 查询 CloudFS 中的数据。

  • 读取 CloudFS 路径下已有的文件。

    val a =sc.textFile("cfs://{Directory}");
    <!-- 从CloudFS路径下读取已有的文件 -->
    a.count
    <!-- 计算读取到的文件 -->
    

  • 写入文件到指定的 CloudFS 路径下。

    a.saveAsTextFile("cfs://{Directory}")