2. 程式人生 > >spark RDD運算元（十一）之RDD Action 儲存操作saveAsTextFile,saveAsSequenceFile,saveAsObjectFile,saveAsHadoopFile 等

spark RDD運算元（十一）之RDD Action 儲存操作saveAsTextFile,saveAsSequenceFile,saveAsObjectFile,saveAsHadoopFile 等

阿新 • • 發佈：2018-12-31

關鍵字:Spark運算元、Spark函式、Spark RDD行動Action、Spark RDD儲存操作、saveAsTextFile、saveAsSequenceFile、saveAsObjectFile,saveAsHadoopFile、saveAsHadoopDataset,saveAsNewAPIHadoopFile、saveAsNewAPIHadoopDataset

saveAsTextFile

def saveAsTextFile(path: String): Unit

def saveAsTextFile(path: String, codec: Class[_ <: CompressionCodec]): Unit

saveAsTextFile用於將RDD以文字檔案的格式儲存到檔案系統中。

codec引數可以指定壓縮的類名。

var rdd1 = sc.makeRDD(1 to 10,2)
scala> rdd1.saveAsTextFile("hdfs://cdh5/tmp/lxw1234.com/") //儲存到HDFS
hadoop fs -ls /tmp/lxw1234.com
Found 2 items
-rw-r--r--   2 lxw1234 supergroup        0 2015-07-10 09:15 /tmp/lxw1234.com/_SUCCESS
-rw-r--r--   2 lxw1234 supergroup        21
 
 2015-07-10 09:15 /tmp/lxw1234.com/part-00000

hadoop fs -cat /tmp/lxw1234.com/part-00000

注意：如果使用rdd1.saveAsTextFile(“file:///tmp/lxw1234.com”)將檔案儲存到本地檔案系統，那麼只會儲存在Executor所在機器的本地目錄。
指定壓縮格式儲存 

rdd1.saveAsTextFile("hdfs://cdh5/tmp/lxw1234.com/",classOf[com.hadoop.compression.lzo.LzopCodec])

hadoop fs -ls /tmp/lxw1234.com
-rw-r--r--   2
 
 lxw1234 supergroup    0 2015-07-10 09:20 /tmp/lxw1234.com/_SUCCESS
-rw-r--r--   2 lxw1234 supergroup    71 2015-07-10 09:20 /tmp/lxw1234.com/part-00000.lzo

hadoop fs -text /tmp/lxw1234.com/part-00000.lzo

saveAsSequenceFile

saveAsSequenceFile用於將RDD以SequenceFile的檔案格式儲存到HDFS上。

用法同saveAsTextFile。

saveAsObjectFile

def saveAsObjectFile(path: String): Unit

saveAsObjectFile用於將RDD中的元素序列化成物件，儲存到檔案中。

對於HDFS，預設採用SequenceFile儲存。 

var rdd1 = sc.makeRDD(1 to 10,2)
rdd1.saveAsObjectFile("hdfs://cdh5/tmp/lxw1234.com/")

hadoop fs -cat /tmp/lxw1234.com/part-00000
SEQ !org.apache.hadoop.io.NullWritable"org.apache.hadoop.io.BytesWritableT

saveAsHadoopFile

def saveAsHadoopFile(path: String, keyClass: Class[], valueClass: Class[], outputFormatClass: Class[_ <: OutputFormat[, ]], codec: Class[_ <: CompressionCodec]): Unit

def saveAsHadoopFile(path: String, keyClass: Class[], valueClass: Class[], outputFormatClass: Class[_ <: OutputFormat[, ]], conf: JobConf = …, codec: Option[Class[_ <: CompressionCodec]] = None): Unit

saveAsHadoopFile是將RDD儲存在HDFS上的檔案中，支援老版本Hadoop API。

可以指定outputKeyClass、outputValueClass以及壓縮格式。

每個分割槽輸出一個檔案。

var rdd1 = sc.makeRDD(Array(("A",2),("A",1),("B",6),("B",3),("B",7)))

import org.apache.hadoop.mapred.TextOutputFormat
import org.apache.hadoop.io.Text
import org.apache.hadoop.io.IntWritable

rdd1.saveAsHadoopFile("/tmp/lxw1234.com/",classOf[Text],classOf[IntWritable],classOf[TextOutputFormat[Text,IntWritable]])

rdd1.saveAsHadoopFile("/tmp/lxw1234.com/",classOf[Text],classOf[IntWritable],classOf[TextOutputFormat[Text,IntWritable]],
                      classOf[com.hadoop.compression.lzo.LzopCodec])

saveAsHadoopDataset

def saveAsHadoopDataset(conf: JobConf): Unit

saveAsHadoopDataset用於將RDD儲存到除了HDFS的其他儲存中，比如HBase。

在JobConf中，通常需要關注或者設定五個引數：

檔案的儲存路徑、key值的class型別、value值的class型別、RDD的輸出格式(OutputFormat)、以及壓縮相關的引數。
##使用saveAsHadoopDataset將RDD儲存到HDFS中 

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import SparkContext._
import org.apache.hadoop.mapred.TextOutputFormat
import org.apache.hadoop.io.Text
import org.apache.hadoop.io.IntWritable
import org.apache.hadoop.mapred.JobConf



var rdd1 = sc.makeRDD(Array(("A",2),("A",1),("B",6),("B",3),("B",7)))
var jobConf = new JobConf()
jobConf.setOutputFormat(classOf[TextOutputFormat[Text,IntWritable]])
jobConf.setOutputKeyClass(classOf[Text])
jobConf.setOutputValueClass(classOf[IntWritable])
jobConf.set("mapred.output.dir","/tmp/lxw1234/")
rdd1.saveAsHadoopDataset(jobConf)

結果：
hadoop fs -cat /tmp/lxw1234/part-00000
A       2
A       1
hadoop fs -cat /tmp/lxw1234/part-00001
B       6
B       3
B       7

##儲存資料到HBASE
HBase建表：

create ‘lxw1234′,{NAME => ‘f1′,VERSIONS => 1},{NAME => ‘f2′,VERSIONS => 1},{NAME => ‘f3′,VERSIONS => 1} 

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import SparkContext._
import org.apache.hadoop.mapred.TextOutputFormat
import org.apache.hadoop.io.Text
import org.apache.hadoop.io.IntWritable
import org.apache.hadoop.mapred.JobConf
import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.mapred.TableOutputFormat
import org.apache.hadoop.hbase.client.Put
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.hbase.io.ImmutableBytesWritable

var conf = HBaseConfiguration.create()
    var jobConf = new JobConf(conf)
    jobConf.set("hbase.zookeeper.quorum","zkNode1,zkNode2,zkNode3")
    jobConf.set("zookeeper.znode.parent","/hbase")
    jobConf.set(TableOutputFormat.OUTPUT_TABLE,"lxw1234")
    jobConf.setOutputFormat(classOf[TableOutputFormat])

    var rdd1 = sc.makeRDD(Array(("A",2),("B",6),("C",7)))
    rdd1.map(x => 
      {
        var put = new Put(Bytes.toBytes(x._1))
        put.add(Bytes.toBytes("f1"), Bytes.toBytes("c1"), Bytes.toBytes(x._2))
        (new ImmutableBytesWritable,put)
      }
    ).saveAsHadoopDataset(jobConf)

##結果：
hbase(main):005:0> scan 'lxw1234'
ROW     COLUMN+CELL                                                                                                
 A       column=f1:c1, timestamp=1436504941187, value=\x00\x00\x00\x02                                              
 B       column=f1:c1, timestamp=1436504941187, value=\x00\x00\x00\x06                                              
 C       column=f1:c1, timestamp=1436504941187, value=\x00\x00\x00\x07                                              
3 row(s) in 0.0550 seconds

注意：儲存到HBase，執行時候需要在SPARK_CLASSPATH中加入HBase相關的jar包。

saveAsNewAPIHadoopFile

def saveAsNewAPIHadoopFile[F <: OutputFormat[K, V]](path: String)(implicit fm: ClassTag[F]): Unit

def saveAsNewAPIHadoopFile(path: String, keyClass: Class[], valueClass: Class[], outputFormatClass: Class[_ <: OutputFormat[, ]], conf: Configuration = self.context.hadoopConfiguration): Unit

saveAsNewAPIHadoopFile用於將RDD資料儲存到HDFS上，使用新版本Hadoop API。

用法基本同saveAsHadoopFile。

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import SparkContext._
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat
import org.apache.hadoop.io.Text
import org.apache.hadoop.io.IntWritable

var rdd1 = sc.makeRDD(Array(("A",2),("A",1),("B",6),("B",3),("B",7)))
rdd1.saveAsNewAPIHadoopFile("/tmp/lxw1234/",classOf[Text],classOf[IntWritable],classOf[TextOutputFormat[Text,IntWritable]])

saveAsNewAPIHadoopDataset

def saveAsNewAPIHadoopDataset(conf: Configuration): Unit

作用同saveAsHadoopDataset,只不過採用新版本Hadoop API。

以寫入HBase為例：

HBase建表：

create ‘lxw1234′,{NAME => ‘f1′,VERSIONS => 1},{NAME => ‘f2′,VERSIONS => 1},{NAME => ‘f3′,VERSIONS => 1}

完整的Spark應用程式： 

package com.lxw1234.test

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import SparkContext._
import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.mapreduce.Job
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.client.Result
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.hbase.client.Put

object Test {
  def main(args : Array[String]) {
   val sparkConf = new SparkConf().setMaster("spark://lxw1234.com:7077").setAppName("lxw1234.com")
   val sc = new SparkContext(sparkConf);
   var rdd1 = sc.makeRDD(Array(("A",2),("B",6),("C",7)))

    sc.hadoopConfiguration.set("hbase.zookeeper.quorum ","zkNode1,zkNode2,zkNode3")
    sc.hadoopConfiguration.set("zookeeper.znode.parent","/hbase")
    sc.hadoopConfiguration.set(TableOutputFormat.OUTPUT_TABLE,"lxw1234")
    var job = new Job(sc.hadoopConfiguration)
    job.setOutputKeyClass(classOf[ImmutableBytesWritable])
    job.setOutputValueClass(classOf[Result])
    job.setOutputFormatClass(classOf[TableOutputFormat[ImmutableBytesWritable]])

    rdd1.map(
      x => {
        var put = new Put(Bytes.toBytes(x._1))
        put.add(Bytes.toBytes("f1"), Bytes.toBytes("c1"), Bytes.toBytes(x._2))
        (new ImmutableBytesWritable,put)
      }    
    ).saveAsNewAPIHadoopDataset(job.getConfiguration)

    sc.stop()   
  }
}

注意：儲存到HBase，執行時候需要在SPARK_CLASSPATH中加入HBase相關的jar包。

相關推薦

spark RDD運算元（十一）之RDD Action 儲存操作saveAsTextFile,saveAsSequenceFile,saveAsObjectFile,saveAsHadoopFile 等

關鍵字:Spark運算元、Spark函式、Spark RDD行動Action、Spark RDD儲存操作、saveAsTextFile、saveAsSequenceFile、saveAsObjectFile,saveAsHadoopFile、saveAsHa

Python爬蟲從入門到放棄（十一）之 Scrapy框架整體的一個了解

object 定義 roc encoding eth obi pipe pos 等等 這裏是通過爬取伯樂在線的全部文章為例子，讓自己先對scrapy進行一個整理的理解 該例子中的詳細代碼會放到我的github地址：https://github.com/pythonsite/

linux（十一）之初始化文件

fun ps1 史記 intern header 例如 internet $path 名稱 前面寫了很多linux的知識，其實很多都是命令的，所以要去多多的練習才能學的更好，加油為了好工作。 要麽現在懶惰，未來討飯。要麽現在努力，未來愜意。 一、初始化文件概述 　　1

python學習筆記（十一）之函數

last 函數返回 traceback keep disco show 全局變量 not 默認參數 牛刀小試： 　　定義一個無參函數 1 >>> def myFirstFunc(): 2 ... print("Hello python

Java基礎入門（十一）之基本數據包裝類以及簡單轉換

數據包 intvalue nbsp 1.5 lse false 永遠 ring jdk 一、 基本數據類型包裝類 引用數據類型一般為基本數據類型首字母大寫，除了int 、char,其中int的引用數據類型類Integer,char的引用數據類型為Character 關

Openstack入門篇（十一）之neutron服務（控制節點）的部署與測試

驅動 username 禁止 普通 虛擬主機 group 查看 禁用 poi 1.Neutron的介紹 　　Neutron 為整個 OpenStack 環境提供網絡支持，包括二層交換，三層路由，負載均衡，防火墻和 *** 等。Neutron 提供了一個靈

Kubernetes學習之路（十一）之資源清單定義

map latest dem kubectl 服務發現 bject 均衡 ima limit 一、Kubernetes常用資源 以下列舉的內容都是 kubernetes 中的 Object，這些對象都可以在 yaml 文件中作為一種 API 類型來配置。 類別 名稱

《大話設計模式》Java程式碼示例（十一）之抽象工廠模式

抽象工廠模式（Abstract Factory）：提供一個建立一系列相關或相互依賴物件的介面，而無需指定它們具體的類。 package abstractfactory; /** * 抽象工廠方法(Abstract Factory) * 使用者實體類 */ p

百度大腦人臉識別深度驗證與思考（十一）之動態實時

前言 我已經厭倦了靜態圖片的識別，那些技術對我已經沒有了挑戰性。今天我們就來看看動態實時的深度識別表現如何。 攝像頭 央視 我們直接採集央視rtmp推流地址的視訊，直接進行人臉識別和即時

web開發（十一）之Hibernate關聯關係配置

寫在前面 Hibernate中關係的對映共有以下四種：一對多、多對一、一對一、多對多這四種。 一對多單向關聯 xml對映 一對多即在A表中的每一條資料都會與B表中的n條有關聯；在這種情況下一般都是在B表新增一個欄位用來當作外來鍵與A表中的主鍵相關聯。而這種

超級終端下必會的命令大全（十一）之功能最強在的編輯器——vi

功能最強在的編輯器——vivi是所有UNIX系統都會提供的螢幕編輯器，它提供了一個視窗裝置，通過它可以編輯檔案。當然，對UNIX系統略有所知的人，或多或少都覺得vi超級難用，但vi是最基本的編輯器，所以希望讀者能好好把它學起來，以後在UNIX世界裡必將暢行無阻、遊刃有餘，因

Ethercat解析（十一）之分佈時鐘

驅動程式碼中，同步時鐘涉及到如下幾個概念： ⑴ 本地時鐘：每一個支援DC的從站都有一個納秒級解析度的本地時鐘暫存器。 每次從機上電，則暫存器從0開始計時，這就意味著不同的從機因為上電開機的時間不同而本地時鐘也會有差異，所以需要對從機的本地時鐘對比參考時鐘

React Native入門（十一）之螢幕適配

準備 首先，我們在官方文件寬度和高度一節可以知道，RN中單位是dp，這個跟Android中的單位是一致的！ 官網中： A dp is equal to one physical pixel on a screen with a density of 1

linux（十一）之初始化檔案

前面寫了很多linux的知識，其實很多都是命令的，所以要去多多的練習才能學的更好，加油為了好工作。 要麼現在懶惰，未來討飯。要麼現在努力，未來愜意。 一、初始化檔案概述 　　1.1、概述 　　系統初始化檔案是給系統做配置資訊的，最重要的系統資訊是環境變數的配置。 　　初始化

【原創】Linux虛擬化KVM-Qemu分析（十一）之virtqueue

# 背景 - `Read the fucking source code!` --By 魯迅 - `A picture is worth a thousand words.` --By 高爾基 說明： 1. KVM版本：5.9.1 2. QEMU版本：5.0.0 3. 工具：Source Insight

Java併發程式設計（十一）Java中的原子操作類

一、原子操作類簡介 JDK1.5開始提供了java.util.concurrent.atomic包，其中有一系列用法簡單、效能高效、可以執行緒安全更新變數的原子操作類，目前（JDK1.7）大概有這麼些： 二、原子操作類實現原理 以AtomicInteger為例看下原始碼，其中的兩個

資料結構（十一）迴圈佇列的基本操作----6個基本操作

//順序佇列存在一個問題 ---假溢位現象，為了解決這個問題，提出了迴圈佇列 //迴圈佇列中存在隊空和隊滿條件一樣的情況，因此提出了犧牲一個空間的方法 //迴圈佇列的基本操作 #include &

spark RDD運算元（十）之PairRDD的Action操作countByKey, collectAsMap

countByKey def countByKey(): Map[K, Long] 以RDD{(1, 2),(2,4),(2,5), (3, 4),(3,5), (3, 6)}為例 rdd.cou

Spark學習之路 （十一）SparkCore的調優之Spark內存模型

精準 規模 memory 此外 結構定義 申請 管理方式 存儲 內部 摘抄自：https://www.ibm.com/developerworks/cn/analytics/library/ba-cn-apache-spark-memory-management/index

Spark項目之電商用戶行為分析大數據平臺之（十一）JSON及FASTJSON

附加 處理 用戶行為分析 基於 大數據平臺 base IT 為什麽 init 一、概述 JSON的全稱是”JavaScript Object Notation”，意思是JavaScript對象表示法，它是一種基於文本，獨立於語言的輕量級數據交換格式