Pyspark獲取并處理RDD數(shù)據(jù)代碼實(shí)例

彈性分布式數(shù)據(jù)集（RDD）是一組不可變的JVM對(duì)象的分布集，可以用于執(zhí)行高速運(yùn)算，它是Apache Spark的核心。

在pyspark中獲取和處理RDD數(shù)據(jù)集的方法如下：

1. 首先是導(dǎo)入庫(kù)和環(huán)境配置（本測(cè)試在linux的pycharm上完成）

import os
from pyspark import SparkContext, SparkConf
from pyspark.sql.session import SparkSession
os.environ["PYSPARK_PYTHON"]="/usr/bin/python3"
conf = SparkConf().setAppName('test_rdd')
sc = SparkContext('local', 'test', conf=conf)
spark = SparkSession(sc)

2. 然后，提供hdfs分區(qū)數(shù)據(jù)的路徑或者分區(qū)表名

txt_File = r"hdfs://host:port/apps/hive/warehouse/數(shù)據(jù)庫(kù)名.db/表名/分區(qū)名/part-m-00029.deflate" # part-m-00029.deflate
# txt_File = r"hdfs://host:port/apps/hive/warehouse/數(shù)據(jù)庫(kù)名.db/表名" # hive table

3. sc.textFile進(jìn)行讀取，得到RDD格式數(shù)據(jù)<還可以用 spark.sparkContext.parallelize(data) 來(lái)獲取RDD數(shù)據(jù)>，參數(shù)中還可設(shè)置數(shù)據(jù)被劃分的分區(qū)數(shù)

txt_ = sc.textFile(txt_File)

4. 基本操作：

• type(txt_)：顯示數(shù)據(jù)類(lèi)型，這時(shí)屬于 'pyspark.rdd.RDD'
• txt_.first()：獲取第一條數(shù)據(jù)
• txt_.take(2)：獲取前2條數(shù)據(jù)，形成長(zhǎng)度為2的list
• txt_.take(2)[1].split('\1')[1]：表示獲取前兩條中的第[1]條數(shù)據(jù)（也就是第2條，因?yàn)閜ython的索引是從0開(kāi)始的），并以 '\1'字符分隔開(kāi)（這要看你的表用什么作為分隔符的），形成list，再獲取該list的第2條數(shù)據(jù)
• txt_.map(lambda x:x.split('\1'))：使用lambda函數(shù)和map函數(shù)快速處理每一行數(shù)據(jù)，這里表示將每一行以 '\1'字符分隔開(kāi)，每一行返回一個(gè)list；此時(shí)數(shù)據(jù)結(jié)構(gòu)是：'pyspark.rdd.PipelinedRDD'
• txt_.map(lambda x:(x, x.split('\1'))).filter(lambda y:y[0].startswith('北京'))：表示在返回 (x, x.split('\1')) 后，進(jìn)行篩選filter，獲取其中以 '北京' 開(kāi)頭的行，并按照相同格式 （例如，這里是(x, x.split('\1'))格式，即原數(shù)據(jù)+分割后的列表數(shù)據(jù)） 返回?cái)?shù)據(jù)
• txt_.collect()：返回所有RDD數(shù)據(jù)元素，當(dāng)數(shù)據(jù)量很大時(shí)謹(jǐn)慎操作
• txt_.toDF()：不能直接轉(zhuǎn)成DataFrame格式，需要設(shè)置Schema

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持億速云。