antony@notes:~/data-platform$ cat "RDD.md"
RDD
RDD
目錄
[TOC]
認識 Resilient Distributed Datasets
14:50
Resilient This means fault-tolerant. By using RDD lineage graph(DAG), we can recompute missing or damaged partitions due to node failures. 有彈性的運算機制，有部分 Node 下線不需要重新運算
Distributed It means data resides on multiple nodes. 具備多點同時處理資料的特性
Dataset It is nothing but a record of the data you work with. Also, a user can load the dataset externally. For example, JSON file, CSV file, text file or database via JDBC with no specific data structure. 可處理資料型態相對 Hive、Pig 多樣化
資料在處理的過程都會放在記憶體裡面處理，所以處理速度會比放在硬碟處理快
MapReduce 在運算時部分 Node 故障下線，MapReduce 會用剩下的 Node 去重新處理資料
認識 RDD Operations
Transformations Spark Transformation is a function that produces new RDD from the existing RDDs. It takes RDD as input and produces one or more RDD as output. Each time it creates new RDD when we apply any transformation. Thus, the so input RDDs, cannot be changed since RDD are immutable in nature.
Transformations are lazy in nature i.e., they get execute when we call an action. They are not executed immediately. Two most basic type of transformations is a map(), filter().
After the transformation, the resultant RDD is always different from its parent RDD. It can be smaller (e.g. filter, count, distinct, sample), bigger (e.g. flatMap(), union(), Cartesian()) or the same size (e.g. map).
當此指令出現時，不會立即處理分析，會先做記錄
Actions Transformations create RDDs from each other, but when we want to work with the actual dataset, at that point action is performed. When the action is triggered after the result, new RDD is not formed like transformation. Thus, Actions are Spark RDD operations that give non-RDD values. The values of action are stored to drivers or to the external storage system. It brings laziness of RDD into motion.
當此指令出現時，整個資料的分析流程才會開始
Spark在執行的過程當中，會提供一些功能，當我們程式在執行時，他只會
Spark RDD 程式設計
以交付模式執行 spkrdd1.py 分析程式
在 Windows 系統的 cmd.exe 視窗, 執行以下命令
$ ssh bigred@<dta1 ip>
bigred@172.16.119.3's password: bigred
$ echo $'from pyspark import SparkConf
from pyspark import SparkContext
conf = SparkConf()
conf.setAppName(\'spkrdd1\')
sc = SparkContext(conf=conf)
## Transformations動作
lines = sc.parallelize(["hello world", "hi"])
## 宣告這行目前尚未被執行，等等的line會被空格做切割
words = lines.flatMap(lambda line: line.split(\' \'))
## fisrt ，是rdd的action命令
print (words.first())' > spkrdd1.py
[註] 上式的 $ 會將 \' 轉換成 '
$ spark-submit --master yarn spkrdd1.py
hello以交付模式執行 spkrdd2.py 分析程式
$ echo $'from pyspark import SparkConf, SparkContext
conf = SparkConf()
conf.setAppName(\'spkrdd2\')
sc = SparkContext(conf=conf)
words = sc.parallelize (
["scala","java","hadoop","spark","akka","spark vs hadoop","pyspark",
"pyspark and spark"]
)
words_filter = words.filter(lambda x: \'spark\' in x)
## count()，為action命令，執行到這行，上面的word和words_filter才會執行
counts = words_filter.count()
print ("Number of elements filtered -> %i" % (counts))' > spkrdd2.py
$ spark-submit spkrdd2.py
Number of elements filtered -> 4以交付模式執行 spkrdd3.py 分析程式
echo $'from pyspark import SparkConf, SparkContext
conf = SparkConf()
conf.setAppName(\'spkrdd2\')
sc = SparkContext(conf=conf)
words = sc.parallelize (
["scala","java","hadoop","spark","akka","spark vs hadoop","pyspark",
"pyspark and spark"]
)
words_filter = words.filter(lambda x: \'spark\' in x)
counts = words_filter.count()
words_filter.saveAsTextFile("myword")
x = sc.textFile("myword")
print (x.take(counts))' > spkrdd3.py
$ spark-submit spkrdd3.py
['spark', 'spark vs hadoop', 'pyspark', 'pyspark and spark']
$ spark-submit spkrdd3.py
...
Output directory hdfs://dtm1:8020/user/bigred/myword already exists
...
$ hdfs dfs -rm -r myword
Deleted myword
$ spark-submit spkrdd3.py
['spark', 'spark vs hadoop', 'pyspark', 'pyspark and spark']檢測 RDD Partition
bigred@dta1:~$ wget "http://www.oc99.org/dt/dataset/goodword.zip" &>/dev/null
bigred@dta1:~$ unzip goodword.zip
Archive: goodword.zip
inflating: goodword.txt
bigred@dta1:~$ dir goodword.txt
-rw-r--r-- 1 bigred bigred 626M Sep 2 2021 goodword.txt
bigred@dta1:~$ hdfs dfs -put goodword.txt mydataset/626MB的檔案被切成5個block，所以總共要啟動5個task
bigred@dta1:~$ nano goodwords.py
from pyspark import SparkConf, SparkContext
conf = SparkConf()
conf.setAppName('wordcount')
sc = SparkContext(conf=conf)
words = sc.textFile("/user/bigred/mydataset/goodword.txt").flatMap(lambda line: line.split(" "))
print("parallelize : "+str(words.getNumPartitions()))
# count the occurrence of each word
wordCounts = words.map(lambda word: (word, 1)).reduceByKey(lambda a,b:a+b)
wf = wordCounts.coalesce(1)
wf.saveAsTextFile("/user/bigred/mydataset/wordcount")## 未指定一個Excuter多少core
bigred@dta1:~$ time spark-submit goodwords.py
parallelize : 5
real 1m30.657s
user 0m9.530s
sys 0m11.988s
同時開另一個終端機下hls
bigred@dta1:~$ hls
[dtm1]
181104 NameNode
181339 SecondaryNameNode
[dtm2]
48403 ResourceManager
48554 JobHistoryServer
[dtw1]
216720 YarnCoarseGrainedExecutorBackend
204667 NodeManager
181803 DataNode
[dtw2]
26142 DataNode
50014 NodeManager
[dtw3]
50960 NodeManager
26484 DataNode
63191 YarnCoarseGrainedExecutorBackend
63035 ExecutorLauncher
[dta1]
216029 SparkSubmit未指定一個Excuter多少core
bigred@dta1:~$ time spark-submit --num-executors 3 goodwords.py
parallelize : 5
real 1m8.794s
user 0m10.798s
sys 0m5.279s
同時開另一個終端機下hls
bigred@dta1:~$ hls
[dtm1]
181104 NameNode
181339 SecondaryNameNode
[dtm2]
48403 ResourceManager
48554 JobHistoryServer
[dtw1]
218278 YarnCoarseGrainedExecutorBackend
204667 NodeManager
181803 DataNode
[dtw2]
63537 ExecutorLauncher
26142 DataNode
50014 NodeManager
63679 YarnCoarseGrainedExecutorBackend
[dtw3]
50960 NodeManager
26484 DataNode
64838 YarnCoarseGrainedExecutorBackend
[dta1]
218019 SparkSubmit指定一個Excuter 2 core
bigred@dta1:~$ time spark-submit --num-executors 3 --executor-cores 2 goodwords.py
parallelize : 5
real 0m51.977s
user 0m10.105s
sys 0m7.079sSpark SQL
交付模式執行 spksql1.py 分析程式
- json，全名：JavaScript Object Notation Java語言對於物件的描述，他很適合用來表達我們的資料，而這個格式就是json
瀏覽器要運作有4大天王：html + dom + JavaScript + css 由 HTML 描述 DOM 再由 JavaScript 操作並使用 CSS 塑造外觀 以上4個都會可以去當web developer 如果只會html + css 可以去當 web ui（不過需要顏色配置上的天賦），因為主要是在設計整個網站的介面
產生一個json資料檔
$ echo $'{"name":"Michael"}
{"name":"Andy","age":30}
{"name":"Justin","age":19}' > pepole.json一個左大括號與一個右大括號為一筆資料
## 將資料丟到Hdfs檔案系統
bigred@dta1:~$ hdfs dfs -put pepole.json
bigred@dta1:~$ echo $'from pyspark.sql import SparkSession
> spark = SparkSession.builder.getOrCreate() ## 這行能讓我們等一下馬上創造出table
> df=spark.read.json("pepole.json") ## 資料表的schema
> df.show()' > spksql1.py
bigred@dta1:~$ spark-submit spksql1.py
+----+-------+
| age| name|
+----+-------+
|null|Michael|
| 30| Andy|
| 19| Justin|
+----+-------+以交付模式執行 spksql2.py 分析程式
編輯json檔格式的資料
bigred@dta1:~$ echo $'from pyspark.sql import SparkSession
> spark = SparkSession.builder.getOrCreate()
> df=spark.read.json("pepole.json")
> df.printSchema()
>
> df.select(df["name"],df["age"]+1).show()
>
> df.filter(df["age"]>20).show() ' > spksql2.py透過Spark執行剛剛的json檔
bigred@dta1:~$ spark-submit spksql2.py
root
|-- age: long (nullable = true)
|-- name: string (nullable = true)
+-------+---------+
| name|(age + 1)|
+-------+---------+
|Michael| null|
| Andy| 31|
| Justin| 20|
+-------+---------+
+---+----+
|age|name|
+---+----+
| 30|Andy|
+---+----+交付模式執行 spksql3.py 分析程式
編輯一個json檔
bigred@dta1:~$ echo $'from pyspark.sql import SparkSession
>
> spark = SparkSession.builder.getOrCreate()
> df=spark.read.json("pepole.json")
>
> df.groupBy("age").count().show()
> df.sort(df["age"].desc()).show()' > spksql3.py透過Spark執行剛剛的json檔
bigred@dta1:~$ spark-submit --master yarn spksql3.py
+----+-----+
| age|count|
+----+-----+
| 19| 1|
|null| 1|
| 30| 1|
+----+-----+
+----+-------+
| age| name|
+----+-------+
| 30| Andy|
| 19| Justin|
|null|Michael|
+----+-------+以交付模式執行 spksql4.py 分析程式
編輯一個json檔
bigred@dta1:~$ echo $'from pyspark.sql import SparkSession
> spark = SparkSession.builder.getOrCreate()
> spark.conf.set("spark.sql.shuffle.partitions", 1)
>
> df=spark.read.json("pepole.json")
>
> dfs=df.sort(df["age"].desc())
> dfs.write.format("json").save("pepole_sort.json")' > spksql4.py執行與檢查
bigred@dta1:~$ spark-submit spksql4.py
bigred@dta1:~$ hdfs dfs -ls pepole_sort.json
Found 2 items
-rw-r--r-- 2 bigred bigboss 0 2022-05-17 01:55 pepole_sort.json/_SUCCESS
-rw-r--r-- 2 bigred bigboss 71 2022-05-17 01:55 pepole_sort.json/part-00000-fa7c1b97-e38b-4805-bd34-d2bf15a78a1d-c000.json
bigred@dta1:~$ hdfs dfs -cat pepole_sort.json/part-00000-fa7c1b97-e38b-4805-bd34-d2bf15a78a1d-c000.json
{"age":30,"name":"Andy"}
{"age":19,"name":"Justin"}
{"name":"Michael"}以交付模式執行 spksql5.py 分析程式
編輯一個json檔
$ echo $'from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
df=spark.read.json("pepole.json")
df.createOrReplaceTempView("pepole_table")
df.cache()
resultsDF = spark.sql("SELECT * FROM pepole_table")
resultsDF.show(2)' > spksql5.py可以直接飆SQL Command
執行
$ spark-submit --master yarn spksql5.py
+----+-------+
| age| name|
+----+-------+
|null|Michael|
| 30| Andy|
+----+-------+
only showing top 2 rows==研究Spark看文章看到spark = SparkSession.builder.getOrCreate()這行，再去閱讀那篇文章==
GroupLens
整理與檢視 u.data 資料集
bigred@dta1:~$ wget 'https://files.grouplens.org/datasets/movielens/ml-100k.zip' &>/dev/null
bigred@dta1:~$ unzip ml-100k.zip &>/dev/null
bigred@dta1:~$ head -n 3 ml-100k/u.data
196 242 3 881250949
186 302 3 891717742
22 377 1 878887116
## user id | item id | rating | timestamp.
上傳 u.data 至 HDFS
bigred@dta1:~$ hdfs dfs -mkdir -p mydataset/ml; hdfs dfs -put ml-100k/u.data mydataset/ml/撰寫與執行 myML01.py
$ nano myML01.py
from pyspark import SparkConf, SparkContext
conf = SparkConf()
conf.setAppName('spkml')
sc = SparkContext(conf=conf)
## 資料讀取到 Executor 的記憶體中
rdd = sc.textFile("/dataset/ml/u.data")
## '\t' 代表 Tab，[:3] 代表只取三個欄位
raw_ratings = rdd.map(lambda line: line.split("\t")[:3])
## 計數資料數量
print (raw_ratings.count())
## 取兩筆資料顯示
print (raw_ratings.take(2))
## 取出參與者筆數（不重複）
num_users = raw_ratings.map(lambda x: x[0]).distinct().count()
## 取出電影筆數（不重複）
num_movies = raw_ratings.map(lambda x: x[1]).distinct().count()
## 顯示上兩個數值
print (num_users,num_movies)bigred@dta1:~$ spark-submit myML01.py
100000
[['196', '242', '3'], ['186', '302', '3']]
943 1682撰寫 myML-2m.py
$ nano myML-2m.py
from pyspark import SparkConf, SparkContext
from pyspark.mllib.recommendation import ALS
conf = SparkConf()
conf.setAppName('spkml')
sc = SparkContext(conf=conf)
rdd = sc.textFile("mydataset/ml/u.data")
raw_ratings = rdd.map(lambda line: line.split("\t")[:3])
# alternating least squares (ALS) algorithm
model = ALS.train(raw_ratings, 10, 10, 0.01)
# User ID 為 10, 有興趣的 5 部 Movie
print (model.recommendProducts(10,5))
# User ID 為 10, 對 Movie 793 的興趣
print (model.predict(10,793))
# Movie 200 有興趣的 5 位 User
print (model.recommendUsers(product=200,num=5))bigred@dta1:~$ spark-submit myML-2m.py
[Rating(user=10, product=1643, rating=5.256996914990797), Rating(user=10, product=1131, rating=5.14216181023412), Rating(user=10, product=1449, rating=5.124614615562111), Rating(user=10, product=318, rating=5.0720185300899265), Rating(user=10, product=1169, rating=5.0102908782616735)]
3.336593846280392
[Rating(user=86, product=200, rating=5.889509864929162), Rating(user=547, product=200, rating=5.729359529809475), Rating(user=153, product=200, rating=5.622134384555626), Rating(user=61, product=200, rating=5.618051883323263), Rating(user=762, product=200, rating=5.5938395090862105)]練習實作
bigred@dta1:~$ hdfs dfs -ls -h mydataset/ml-25m/
Found 1 items
-rw-r--r-- 2 bigred bigboss 646.8 M 2022-05-17 03:09 mydataset/ml-25m/ratings.csv