溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
學習spark任何技術之前，請先正確理解spark，可以參考：正確理解spark
以下對RDD的三種創建方式、單類型RDD基本的transformation api、采樣Api以及pipe操作進行了java api方面的闡述
一、RDD的三種創建方式
從穩定的文件存儲系統中創建RDD，比如local fileSystem或者hdfs等，如下：
//從hdfs文件中創建
JavaRDD<String> textFileRDD = sc.textFile("hdfs://master:9999/users/hadoop-twq/word.txt");
//從本地文件系統的文件中，注意file:后面肯定是至少三個///，四個也行，不能是兩個
//如果指定第二個參數的話，表示創建的RDD的最小的分區數，如果文件分塊的數量大于指定的分區
//數的話則已文件的分塊數量為準
JavaRDD<String> textFileRDD = sc.textFile("hdfs://master:9999/users/hadoop-twq/word.txt" 2 );2. 可以經過transformation api從一個已經存在的RDD上創建一個新的RDD，以下是map這個轉換api
JavaRDD<String> mapRDD = textFileRDD.map(new Function<String, String>() {
@Override
public String call(String s) throws Exception {
return s + "test";
}
});
System.out.println("mapRDD = " + mapRDD.collect());3. 從一個內存中的列表數據創建一個RDD，可以指定RDD的分區數，如果不指定的話，則取所有Executor的所有cores數量
//創建一個單類型的JavaRDD
JavaRDD<Integer> integerJavaRDD = sc.parallelize(Arrays.asList(1, 2, 3, 3, 4), 2);
System.out.println("integerJavaRDD = " + integerJavaRDD.glom().collect());
//創建一個單類型且類型為Double的JavaRDD
JavaDoubleRDD doubleJavaDoubleRDD = sc.parallelizeDoubles(Arrays.asList(2.0, 3.3, 5.6));
System.out.println("doubleJavaDoubleRDD = " + doubleJavaDoubleRDD.collect());
//創建一個key-value類型的RDD
import scala.Tuple2;
JavaPairRDD<String, Integer> javaPairRDD = sc.parallelizePairs(Arrays.asList(new Tuple2("test", 3), new Tuple2("kkk", 3)));
System.out.println("javaPairRDD = " + javaPairRDD.collect());注：對于第三種情況，scala中還提供了makeRDD api，這個api可以指定創建RDD每一個分區所在的機器，這個api的原理詳見spark core RDD scala api中
二、單類型RDD基本的transformation api
先基于內存中的數據創建一個RDD
JavaRDD<Integer> integerJavaRDD = sc.parallelize(Arrays.asList(1, 2, 3, 3), 2);
map操作，表示對integerJavaRDD的每一個元素應用我們自定義的函數接口，如下是將每一個元素加1：
JavaRDD<Integer> mapRDD = integerJavaRDD.map(new Function<Integer, Integer>() {
@Override
public Integer call(Integer element) throws Exception {
return element + 1;
}
});
//結果：[2, 3, 4, 4]
System.out.println("mapRDD = " + mapRDD.collect());需要注意的是，map操作可以返回與RDD不同類型的數據，如下，返回一個自定義的User對象：
public class User implements Serializable {
private String userId;
private Integer amount;
public User(String userId, Integer amount) {
this.userId = userId;
this.amount = amount;
}
//getter setter....
@Override
public String toString() {
return "User{" +
"userId='" + userId + '\'' +
", amount=" + amount +
'}';
}
}
JavaRDD<User> userJavaRDD = integerJavaRDD.map(new Function<Integer, User>() {
@Override
public User call(Integer element) throws Exception {
if (element < 3) {
return new User("小于3", 22);
} else {
return new User("大于3", 23);
}
}
});
//結果：[User{userId='小于3', amount=22}, User{userId='小于3', amount=22}, User{userId='大于3', amount=23}, User{userId='大于3', amount=23}]
System.out.println("userJavaRDD = " + userJavaRDD.collect());2. flatMap操作，對integerJavaRDD的每一個元素應用我們自定義的FlatMapFunction，這個函數的輸出是一個數據列表，flatMap會對這些輸出的數據列表進行展平
JavaRDD<Integer> flatMapJavaRDD = integerJavaRDD.flatMap(new FlatMapFunction<Integer, Integer>() {
@Override
public Iterator<Integer> call(Integer element) throws Exception {
//輸出一個list，這個list里的元素是0到element
List<Integer> list = new ArrayList<>();
int i = 0;
while (i <= element) {
list.add(i);
i++;
}
return list.iterator();
}
});
//結果： [0, 1, 0, 1, 2, 0, 1, 2, 3, 0, 1, 2, 3]
System.out.println("flatMapJavaRDD = " + flatMapJavaRDD.collect());3. filter操作，對integerJavaRDD的每一個元素應用我們自定義的過濾函數，過濾掉我們不需要的元素，如下，過濾掉不等于1的元素：
JavaRDD<Integer> filterJavaRDD = integerJavaRDD.filter(new Function<Integer, Boolean>() {
@Override
public Boolean call(Integer integer) throws Exception {
return integer != 1;
}
});
//結果為：[2, 3, 3]
System.out.println("filterJavaRDD = " + filterJavaRDD.collect());4. glom操作，查看integerJavaRDD每一個分區對應的元素數據
JavaRDD<List<Integer>> glomRDD = integerJavaRDD.glom();
//結果： [[1, 2], [3, 3]]， 說明integerJavaRDD有兩個分區，第一個分區中有數據1和2,第二個分區中有數據3和3
System.out.println("glomRDD = " + glomRDD.collect());5. mapPartitions操作，對integerJavaRDD的每一個分區的數據應用我們自定義的函數接口方法，假設我們需要為每一個元素加上一個初始值，而這個初始值的獲取又是非常耗時的，這個時候用mapPartitions會有非常大的優勢，如下：
//這是一個初始值獲取的方法，是一個比較耗時的方法
public static Integer getInitNumber(String source) {
System.out.println("get init number from " + source + ", may be take much time........");
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 1;
}
JavaRDD<Integer> mapPartitionTestRDD = integerJavaRDD.mapPartitions(new FlatMapFunction<Iterator<Integer>, Integer>() {
@Override
public Iterator<Integer> call(Iterator<Integer> integerIterator) throws Exception {
//每一個分區獲取一次初始值，integerJavaRDD有兩個分區，那么會調用兩次getInitNumber方法
//所以對應需要初始化的比較耗時的操作，比如初始化數據庫的連接等，一般都是用mapPartitions來為對每一個分區初始化一次，而不要去使用map操作
Integer initNumber = getInitNumber("mapPartitions");
List<Integer> list = new ArrayList<>();
while (integerIterator.hasNext()) {
list.add(integerIterator.next() + initNumber);
}
return list.iterator();
}
});
//結果為： [2, 3, 4, 4]
System.out.println("mapPartitionTestRDD = " + mapPartitionTestRDD.collect());
JavaRDD<Integer> mapInitNumberRDD = integerJavaRDD.map(new Function<Integer, Integer>() {
@Override
public Integer call(Integer integer) throws Exception {
//遍歷每一個元素的時候都會去獲取初始值，這個integerJavaRDD含有4個元素，那么這個getInitNumber方法會被調用4次，嚴重的影響了時間，不如mapPartitions性能好
Integer initNumber = getInitNumber("map");
return integer + initNumber;
}
});
//結果為：[2, 3, 4, 4]
System.out.println("mapInitNumberRDD = " + mapInitNumberRDD.collect());6. mapPartitionsWithIndex操作，對integerJavaRDD的每一個分區的數據應用我們自定義的函數接口方法，在應用函數接口方法的時候帶上了分區信息，即知道你當前處理的是第幾個分區的數據
JavaRDD<Integer> mapPartitionWithIndex = integerJavaRDD.mapPartitionsWithIndex(new Function2<Integer, Iterator<Integer>, Iterator<Integer>>() {
@Override
public Iterator<Integer> call(Integer partitionId, Iterator<Integer> integerIterator) throws Exception {
//partitionId表示當前處理的第幾個分區的信息
System.out.println("partition id = " + partitionId);
List<Integer> list = new ArrayList<>();
while (integerIterator.hasNext()) {
list.add(integerIterator.next() + partitionId);
}
return list.iterator();
}
}, false);
//結果 [1, 2, 4, 4]
System.out.println("mapPartitionWithIndex = " + mapPartitionWithIndex.collect());三、采樣Api
先基于內存中的數據創建一個RDD
JavaRDD<Integer> listRDD = sc.parallelize(Arrays.asList(1, 2, 3, 3), 2);
sample
//第一個參數為withReplacement
//如果withReplacement=true的話表示有放回的抽樣，采用泊松抽樣算法實現
//如果withReplacement=false的話表示無放回的抽樣，采用伯努利抽樣算法實現
//第二個參數為：fraction，表示每一個元素被抽取為樣本的概率，并不是表示需要抽取的數據量的因子
//比如從100個數據中抽樣，fraction=0.2，并不是表示需要抽取100 * 0.2 = 20個數據，
//而是表示100個元素的被抽取為樣本概率為0.2;樣本的大小并不是固定的，而是服從二項分布
//當withReplacement=true的時候fraction>=0
//當withReplacement=false的時候 0 < fraction < 1
//第三個參數為：reed表示生成隨機數的種子，即根據這個reed為rdd的每一個分區生成一個隨機種子
JavaRDD<Integer> sampleRDD = listRDD.sample(false, 0.5, 100);
//結果： [1, 3]
System.out.println("sampleRDD = " + sampleRDD.collect());2. randomSplit
//按照權重對RDD進行隨機抽樣切分，有幾個權重就切分成幾個RDD
//隨機抽樣采用伯努利抽樣算法實現, 以下是有兩個權重，則會切成兩個RDD
JavaRDD<Integer>[] splitRDDs = listRDD.randomSplit(new double[]{0.4, 0.6});
//結果為2
System.out.println("splitRDDs.length = " + splitRDDs.length);
//結果為[2, 3] 結果是不定的
System.out.println("splitRDD(0) = " + splitRDDs[0].collect());
//結果為[1, 3] 結果是不定的
System.out.println("splitRDD(1) = " + splitRDDs[1].collect());3. takeSample
//隨機抽樣指定數量的樣本數據 //第一個參數為withReplacement //如果withReplacement=true的話表示有放回的抽樣，采用泊松抽樣算法實現 //如果withReplacement=false的話表示無放回的抽樣，采用伯努利抽樣算法實現 //第二個參數指定多少，則返回多少個樣本數 結果為[2, 3] System.out.println(listRDD.takeSample(false, 2));
4. 分層采樣，對key-value類型的RDD進行采樣
//創建一個key value類型的RDD
import scala.Tuple2;
JavaPairRDD<String, Integer> javaPairRDD =
sc.parallelizePairs(Arrays.asList(new Tuple2("test", 3),
new Tuple2("kkk", 3), new Tuple2("kkk", 3)));
//定義每一個key的采樣因子
Map<String, Double> fractions = new HashMap<>();
fractions.put("test", 0.5);
fractions.put("kkk", 0.4);
//對每一個key進行采樣
//結果為 [(test,3), (kkk,3)]
//sampleByKey 并不對過濾全量數據，因此只得到近似值
System.out.println(javaPairRDD.sampleByKey(true, fractions).collect());
//結果為 [(test,3), (kkk,3)]
//sampleByKeyExtra 會對全量數據做采樣計算，因此耗費大量的計算資源，但是結果會更準確。
System.out.println(javaPairRDD.sampleByKeyExact(true, fractions).collect());抽樣的原理詳細可以參考：spark core RDD api。這些原理性的東西用文字不太好表述
四、pipe，表示在RDD執行流中的某一步執行其他的腳本，比如python或者shell腳本等
JavaRDD<String> dataRDD = sc.parallelize(Arrays.asList("hi", "hello", "how", "are", "you"), 2);
//啟動echo.py需要的環境變量
Map<String, String> env = new HashMap<>();
env.put("env", "envtest");
List<String> commands = new ArrayList<>();
commands.add("python");
//如果是在真實的spark集群中，那么要求echo.py在集群的每一臺機器的同一個目錄下面都要有
commands.add("/Users/tangweiqun/spark/source/spark-course/spark-rdd-java/src/main/resources/echo.py");
JavaRDD<String> result = dataRDD.pipe(commands, env);
//結果為： [slave1-hi-envtest, slave1-hello-envtest, slave1-how-envtest, slave1-are-envtest, slave1-you-envtest]
System.out.println(result.collect());echo.py的內容如下：
import sys
import os
#input = "test"
input = sys.stdin
env_keys = os.environ.keys()
env = ""
if "env" in env_keys:
env = os.environ["env"]
for ele in input:
output = "slave1-" + ele.strip('\n') + "-" + env
print (output)
input.close對于pipe的原理，以及怎么實現的，參考：spark core RDD api，這個里面還清楚的講述了怎么消除手工將腳本拷貝到每一臺機器中的工作
系統學習spark：
1、[老湯] Spark 2.x 之精講Spark Core：https://edu.51cto.com/sd/88429
2、[老湯]Spark 2.x 之精講Spark SQL專題：https://edu.51cto.com/sd/16f3d
3、[老湯]Scala內功修煉系列專題：https://edu.51cto.com/sd/8e85b
4、[老湯]Spark 2.x之精講Spark Streamig：https://edu.51cto.com/sd/8c525
5、[老湯]Spark 2.x精講套餐：https://edu.51cto.com/sd/ff9a4
6、從Scala到Spark 2.x專題：https://edu.51cto.com/sd/d72af
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
