一、RDD#flatMap 方法

1、RDD#flatMap 方法引入

RDD#map 方法 可以 將 RDD 中的數(shù)據(jù)元素 逐個進行處理 , 處理的邏輯 需要用外部 通過 參數(shù)傳入 map 函數(shù) ;

RDD#flatMap 方法 是 在 RDD#map 方法 的基礎(chǔ)上 , 增加了 " 解除嵌套 " 的作用 ;

RDD#flatMap 方法 也是 接收一個 函數(shù) 作為參數(shù) , 該函數(shù)被應(yīng)用于 RDD 中的每個元素及元素嵌套的子元素 , 并返回一個 新的 RDD 對象 ;

2、解除嵌套

解除嵌套 含義 : 下面的的 列表 中 , 每個元素 都是一個列表 ;

lst = [[1, 2], [3, 4, 5], [6, 7, 8]]

如果將上述 列表 解除嵌套 , 則新的 列表 如下 :

lst = [1, 2, 3, 4, 5, 6, 7, 8]

RDD#flatMap 方法 先對 RDD 中的 每個元素 進行處理 , 然后再 將 計算結(jié)果展平放到一個新的 RDD 對象中 , 也就是 解除嵌套 ;

這樣 原始 RDD 對象 中的 每個元素 , 都對應(yīng) 新 RDD 對象中的若干元素 ;

3、RDD#flatMap 語法說明

RDD#flatMap 語法說明 :

newRDD = oldRDD.flatMap(lambda x: [element1, element2, ...])

舊的 RDD 對象 oldRDD 中 , 每個元素應(yīng)用一個 lambda 函數(shù) , 該函數(shù)返回多個元素 , 返回的多個元素就會被展平放入新的 RDD 對象 newRDD 中 ;

代碼示例 :

# 將 字符串列表 轉(zhuǎn)為 RDD 對象
rdd = sparkContext.parallelize(["Tom 18", "Jerry 12", "Jack 21"])

# 應(yīng)用 map 操作，將每個元素 按照空格 拆分
rdd2 = rdd.flatMap(lambda element: element.split(" "))

二、代碼示例 - RDD#flatMap 方法

代碼示例 :

"""
PySpark 數(shù)據(jù)處理
"""

# 導(dǎo)入 PySpark 相關(guān)包
from pyspark import SparkConf, SparkContext
# 為 PySpark 配置 Python 解釋器
import os
os.environ['PYSPARK_PYTHON'] = "Y:/002_WorkSpace/PycharmProjects/pythonProject/venv/Scripts/python.exe"

# 創(chuàng)建 SparkConf 實例對象 , 該對象用于配置 Spark 任務(wù)
# setMaster("local[*]") 表示在單機模式下 本機運行
# setAppName("hello_spark") 是給 Spark 程序起一個名字
sparkConf = SparkConf() \
    .setMaster("local[*]") \
    .setAppName("hello_spark")

# 創(chuàng)建 PySpark 執(zhí)行環(huán)境 入口對象
sparkContext = SparkContext(conf=sparkConf)

# 打印 PySpark 版本號
print("PySpark 版本號 : ", sparkContext.version)

# 將 字符串列表 轉(zhuǎn)為 RDD 對象
rdd = sparkContext.parallelize(["Tom 18", "Jerry 12", "Jack 21"])

# 應(yīng)用 map 操作，將每個元素 按照空格 拆分
rdd2 = rdd.flatMap(lambda element: element.split(" "))

# 打印新的 RDD 中的內(nèi)容
print(rdd2.collect())

# 停止 PySpark 程序
sparkContext.stop()

執(zhí)行結(jié)果 :

Y:\002_WorkSpace\PycharmProjects\pythonProject\venv\Scripts\python.exe Y:/002_WorkSpace/PycharmProjects/HelloPython/hello.py
23/07/31 23:02:58 WARN Shell: Did not find winutils.exe: java.io.FileNotFoundException: java.io.FileNotFoundException: HADOOP_HOME and hadoop.home.dir are unset. -see https://wiki.apache.org/hadoop/WindowsProblems
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
23/07/31 23:02:59 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
PySpark 版本號 :  3.4.1
['Tom', '18', 'Jerry', '12', 'Jack', '21']

Process finished with exit code 0

文章來源地址http://www.zghlxwxcb.cn/news/detail-629912.html

到了這里，關(guān)于【Python】PySpark 數(shù)據(jù)計算 ② ( RDD#flatMap 方法 | RDD#flatMap 語法 | 代碼示例 )的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！