<jupyter_start><jupyter_text>4.2 用 Pyspark 建立第一個RDD<jupyter_code>from __future__ import print_function, division<jupyter_output><empty_output><jupyter_text>import pyspark<jupyter_code>import os
import sys 

spark_home = os.environ['SPARK_HOME']
sys.path.insert(0, os.path.join(spark_home, 'python'))
sys.path.insert(0, os.path.join(spark_home, 'python/lib/py4j-0.10.4-src.zip'))
from pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSession<jupyter_output><empty_output><jupyter_text>啟動 spark<jupyter_code>spark = SparkSession.builder.master("local") \
   .appName("test") \
   .enableHiveSupport() \
   .getOrCreate()

sc = spark.sparkContext<jupyter_output><empty_output><jupyter_text>Part1. Create a RDD from sparkContext<jupyter_code>wordsList = ['cat', 'elephant', 'rat', 'rat', 'cat']
wordsRDD = sc.parallelize(wordsList, 4)
# Print out the type of wordsRDD
print(type(wordsRDD))

wordsRDD.take(3)
wordsRDD.collect()<jupyter_output><empty_output><jupyter_text>Part2 Create a Dataframe from hdfs  put data into HDFS<jupyter_code>!ls ../data
!head ../data/NASA_access_log_Jul95_100
!hadoop fs -ls /
!hadoop fs -put ../data/NASA_access_log_Jul95_100 /tmp
!hadoop fs -ls /tmp<jupyter_output><empty_output><jupyter_text>從 HDFS 中讀取資料<jupyter_code>textFromHDFS = spark.read.text("hdfs:///tmp/NASA_access_log_Jul95_100")
print(type(textFromHDFS))
textFromHDFS.head()<jupyter_output><empty_output><jupyter_text>Part3 Read csv format<jupyter_code>!hadoop fs -put ../data/ratings.csv /tmp
!hadoop fs -tail /tmp/ratings.csv
path = "hdfs:///tmp/ratings.csv"
schema = None 
sep = None
header = True
csvDF = spark.read.csv(path = path, schema = schema, sep = sep, header = header)
print(type(csvDF))
print(csvDF)
csvDF.head()<jupyter_output><empty_output><jupyter_text>comapre with read.text<jupyter_code>textDF = spark.read.text(paths = path)
textDF
textDF.head()<jupyter_output><empty_output><jupyter_text>Part4. Read Json file<jupyter_code>!hadoop fs -put ../data/json_example.json /tmp
jsonDF = spark.read.json('hdfs:///tmp/json_example.json')
jsonDF
jsonDF.head()<jupyter_output><empty_output><jupyter_text>Part5. RDD 與 DataFrame 的轉換<jupyter_code>jsonRDD = jsonDF.rdd
print(type(jsonDF))
print(type(jsonRDD))
jsonRDD.head()
jsonRDD.take(1)
jsonDF2 = spark.createDataFrame(jsonRDD)
type(jsonDF2)
jsonDF2.head()
sc.stop()
spark.stop()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1405132519983172, 'fraction_numerical': 0.014724442574673959, 'mean_word_length': 7.1438356164383565, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 41, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '10', 'dir': 'jupyter-structured-clean-dedup', 'id': '336348', 'license': 'MIT', 'n_tokens_mistral': 990, 'n_tokens_neox': 957, 'n_words': 150, 'path': 'spark_core/2_create_rdd.ipynb', 'repo_name': 'bryanyang0528/spark_practice', 'size': '2377'}