<jupyter_start><jupyter_text>PySpark Basics<jupyter_code>from pyspark import SparkContext
from pyspark.sql.session import SparkSession
sc = SparkContext.getOrCreate()
my_list = [1, 2, 3, 4]
squared_my_list = list(map(lambda x: x*x, my_list))
squared_my_list
filtered_my_list = list(filter(lambda x: x%2 !=0, my_list))
filtered_my_list
numbers = list(range(0,100))
numbersRDD = sc.parallelize(numbers)
type(numbersRDD)<jupyter_output><empty_output><jupyter_text>Transformations and Actions<jupyter_code>numRDD = sc.parallelize(my_list)
cubeRDD = numRDD.map(lambda x: x**3)
numbers_all = cubeRDD.collect()

[print(num) for num in numbers_all];
pairRDD = sc.parallelize([(1, 2), (3, 4), (1, 4), (3, 2), (4, 2)])
pairRDD_Reduced = pairRDD.reduceByKey(lambda x, y: x+y)

[print(num) for num in pairRDD_Reduced.collect()];
pairRDD_Reduced_sorted = pairRDD_Reduced.sortByKey(ascending=True)
[print(num) for num in pairRDD_Reduced_sorted.collect()];
total = pairRDD.countByKey()
total
type(total)
total.items()
[print(k, "has", v, "values") for k,v in total.items()];<jupyter_output>1 has 2 values
3 has 2 values
4 has 1 values<jupyter_text>Spark Session<jupyter_code>spark = SparkSession(sc)

sample_list = [("Razi", 39), ("Ali", 37), ("Naqi", 33)]
sampleRDD = sc.parallelize(sample_list)
df_sample = spark.createDataFrame(sampleRDD, schema=["Name", "Age"])
type(df_sample)
df_sample.show()
df_people = spark.read.csv("people.csv", header=True, inferSchema=True)
df_people.show()
df_people.printSchema()
df_people.count()
df_people.columns
df_people_sub = df_people.select("name", "sex", "date of birth")
df_people_sub.show(10)
df_people_nodups = df_people.dropDuplicates()
print("Below Dropping Duplicates", df_people.count())
print("After Dropping Duplicates", df_people_nodups.count())
df_people_females = df_people.filter(df_people.sex == "female")
df_people_males = df_people.filter(df_people.sex == "male")
df_people_females.count()
df_people_males.count()
df_people_sex = df_people.groupBy("sex")
df_people_sex.count().show()
df_people.orderBy("date of birth").show()
df_people = df_people.withColumnRenamed("date of birth", "DOB")
df_people.show()<jupyter_output>+---+---------+-----------------+------+----------+
|_c0|person_id|             name|   sex|       DOB|
+---+---------+-----------------+------+----------+
|  0|      100|   <NAME>|female|1990-08-31|
|  1|      101|    <NAME>|  male|1971-10-14|
|  2|      102|        <NAME>|female|1962-05-24|
|  3|      103|     <NAME>|female|2017-03-10|
|  4|      104|   <NAME>|female|2020-08-05|
|  5|      105|    <NAME>|  male|1999-12-30|
|  6|      106|    <NAME>|  male|1983-05-09|
|  7|      107|     <NAME>|  male|1990-09-13|
|  8|      108| <NAME>|  male|1958-08-08|
|  9|      109|   <NAME>|  male|2005-05-24|
| 10|      110| <NAME>|  male|2002-05-07|
| 11|      111|<NAME>|female|1989-07-13|
| 12|      112|    <NAME>|female|1976-01-19|
| 13|      113|       <NAME>|  male|1970-05-26|
| 14|      114|<NAME>|  male|1980-12-22|
| 15|      115|<NAME>|  male|1973-12-27|
| 16|      1[...]<jupyter_text>Using Spark SQL module<jupyter_code>df_people.createOrReplaceTempView("people")
query = "select name, DOB from people where sex='male' order by DOB"
df_people_names_dob = spark.sql(query)
df_people_names_dob.show()<jupyter_output>+-----------------+----------+
|             name|       DOB|
+-----------------+----------+
|     <NAME>|1903-07-14|
|    <NAME>|1903-11-07|
|     <NAME>|1906-11-04|
|     <NAME>|1907-12-08|
|       <NAME>|1909-07-11|
|      <NAME>|1910-04-14|
|    <NAME>|1911-01-12|
|       <NAME>|1911-03-27|
|  <NAME>|1911-07-06|
|     <NAME>|1911-08-09|
|<NAME>|1912-04-10|
|  <NAME>|1912-07-25|
|      <NAME>|1912-09-09|
|   <NAME>|1914-01-22|
|  <NAME>|1915-03-02|
|     <NAME>|1915-12-03|
|  <NAME>|1916-03-12|
|   <NAME>|1916-05-24|
|    <NAME>|1916-08-16|
|    <NAME>|1916-10-06|
+-----------------+----------+
only showing top 20 rows<jupyter_text>Create RDD from External File<jupyter_code>clusterRDD = sc.textFile("5000_points.txt")
clusterRDD.getNumPartitions()
clusterRDD = sc.textFile("5000_points.txt", 5)
clusterRDD.getNumPartitions()
clusterRDD.take(5)
rdd_split = clusterRDD.map(lambda x: x.split("\t"))
rdd_split.take(5)
rdd_split_int = rdd_split.map(lambda x: [int(x[0]), int(x[1])])
rdd_split_int.take(5)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.2120092378752887, 'fraction_numerical': 0.0953810623556582, 'mean_word_length': 3.7437020810514787, 'pattern_counts': {'":': 0, '<': 54, '<?xml version=': 0, '>': 54, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '540553', 'license': 'MIT', 'n_tokens_mistral': 2047, 'n_tokens_neox': 1763, 'n_words': 289, 'path': 'Untitled.ipynb', 'repo_name': 'raziiq/python-pyspark-basics', 'size': '4593'}