10:23 18/10/24 midterm1 - Databricks

midterm1
([Link]
1

import requests

# GitHub raw file URL

url = '...'

# Download the file

response = [Link](url)
file_content = [Link]

# Save the file to Databricks' DBFS (Databricks File System)

[Link]("/data/...", file_content, overwrite=True)

Wrote 55626540 bytes.

Out[1]: True

display([Link]("/data"))

Table

path name size modificationTime

1 dbfs:/data/email.c… [Link] 55626540 1729219853000

1 row

[Link] (1).html 1/22

10:23 18/10/24 midterm1 - Databricks

# Reading the CSV file from DBFS

df = [Link]("/data/[Link]", header=True, inferSchema=True, sep=',')

[Link]()

|From: [Link]@...| null|

|To: [Link]@enr...| [Link]@enr...|
|\[Link]@en...| [Link]@en...|
|\[Link]@en...| [Link]@enr...|
|\[Link]@...| [Link]@e...|
|\[Link]@e...| [Link]...|
|\[Link]...| [Link]@enro...|
|\[Link]@enr...| [Link]@e...|
|\[Link]...| [Link]@...|
|\[Link]@e...| [Link]@enro...|
|\[Link]@...| [Link]...|
|\[Link]@enr...| [Link]@en...|
|\[Link]...| [Link]@[Link]|
|\[Link]@en...| [Link]@en...|
|\[Link]...| [Link]@en...|
|\[Link]@en...| [Link]@...|
|\[Link]@en...| [Link]@...|
|\[Link]@enr...| [Link]@enro...|
+--------------------+--------------------+
only showing top 20 rows

import os
from [Link] import SparkSession
from [Link] import DataFrame
from [Link] import RDD
from [Link] import Row

sc = [Link] \
.appName("Email Data Processing") \
.getOrCreate()

dataPath = "/data/[Link]"

dataPath = [Link]("DATA_MIDTERM") or dataPath

print(dataPath)

/data/[Link]

[Link] (1).html 2/22

10:23 18/10/24 midterm1 - Databricks

#0.5
def loadAndProcessCsv(filePath: str, spark: SparkSession) -> DataFrame:
'''
This function loads a CSV file into a Spark DataFrame, caches it,
drops rows with null values, and prints the schema. (using option when read to keep format - header, columns)

Args:
file_path (str): Path to the CSV file.
spark (SparkSession): Active Spark session.

Returns:
DataFrame: Processed DataFrame.
'''

data = None
# YOUR CODE HERE
# Step 1: Read the CSV file into a DataFrame, with header and inferSchema options.
data = [Link]("header", "true").option("inferSchema", "true").option("delimiter", ";").csv(filePath)

# Step 2: Cache the DataFrame for efficient reuse.

[Link]()

# Step 3: Drop rows with null values.

data = [Link]()

# Step 4: Print the schema of the DataFrame.

[Link]()

# Dòng raise NotImplementedError() thường được sử dụng trong quá trình phát triển phần mềm để báo hiệu rằng một phần mã chưa
triển khai hoặc hoàn thiện. Khi chương trình gặp phải dòng này, nó sẽ dừng lại và ném ra ngoại lệ (exception) với thông báo
năng chưa được hoàn thành.
# raise NotImplementedError()
return data

data = loadAndProcessCsv(dataPath,sc)

root
|-- file,message: string (nullable = true)

[Link] (1).html 3/22

10:23 18/10/24 midterm1 - Databricks

[Link]

Out[8]: ['file,message']

10

[Link]()

|From: [Link]@...|
|To: [Link]@enr...|
|\[Link]@en...|
|\[Link]@en...|
|\[Link]@...|
|\[Link]@e...|
|\[Link]...|
|\[Link]@enr...|
|\[Link]...|
|\[Link]@e...|
|\[Link]@...|
|\[Link]@enr...|
|\[Link]...|
|\[Link]@en...|
|\[Link]...|
|\[Link]@en...|
|\[Link]@en...|
|\[Link]@enr...|
+--------------------+
only showing top 20 rows

11

[Link] (1).html 4/22

10:23 18/10/24 midterm1 - Databricks

# 0.5đ
def createRDD(data: DataFrame)->RDD[Row]:
'''
This function converts a Spark DataFrame into an RDD of Rows.

Args:
data (DataFrame): Input DataFrame containing data loaded by Spark.

Returns:
RDD[Row]: An RDD containing all Rows from the input DataFrame.
'''
outRDD = None

# YOUR CODE HERE

outRDD = [Link]

# outRDD = [Link](lambda row: Row(**[Link]()))

return outRDD

12

emailRDD = createRDD(data)
assert isinstance(emailRDD, RDD), "createRDD() does not return the correct data type (RDD)"
assert isinstance([Link](),Row), "createRDD() contains elements that are not of type Row"

13

[Link]()

root
|-- file,message: string (nullable = true)

14

[Link] (1).html 5/22

10:23 18/10/24 midterm1 - Databricks

import email
from typing import List, Optional
from [Link] import Row

def splitEmailAddresses(emailString: str) -> List[Optional[str]]:

'''
The function splits a comma-separated string of email addresses into a unique list.

Args:
emailString: A string containing email addresses separated by commas.

Returns:
A list of unique email addresses.
'''
if emailString:
addresses = [Link](',')
uniqueAddresses = list(frozenset(map(lambda x: [Link](), addresses)))
return uniqueAddresses
return []

def extractEmailDetailsFromRawText(rawEmail: str) -> Row:

'''
The function extracts relevant details from a raw email message string.

Args:
rawEmail: A string representing the raw email message.

Returns:
A Row object containing the extracted email details.
'''
emailMessage = email.message_from_string(rawEmail)
emailContentParts = []
for part in [Link]():
if part.get_content_type() == 'text/plain':
[Link](part.get_payload())

emailContent = ''.join(emailContentParts)

fromAddresses = splitEmailAddresses([Link]("From"))
toAddresses = splitEmailAddresses([Link]("To"))
ccEmail = splitEmailAddresses([Link]("Cc"))
return Row(
Date=[Link]("Date"),
From=fromAddresses,
To=toAddresses,
Subject=[Link]("Subject"),
CC=ccEmail,
Content=emailContent
)

[Link] (1).html 6/22

10:23 18/10/24 midterm1 - Databricks

# Extract structured email details from the first email message

firstEmailData = [Link]()
# structuredEmail = extractEmailDetailsFromRawText([Link])
structuredEmail = extractEmailDetailsFromRawText(firstEmailData['file,message'])
structuredEmail

Out[13]: Row(Date=None, From=[], To=[], Subject=None, CC=[], Content='')

15

#0.5đ
def createStructuredEmailRDD(emailRDD: RDD[Row]) -> RDD[Row]:
'''
The function takes an RDD of email messages and converts it into a new RDD containing structured email details.

Args:
emailRDD: An RDD where each Row contains an email message in raw text format.

Returns:
A new RDD where each element is a Row with structured email details such as Date, From, To, Subject, CC, and Content.
'''
structuredEmailRDD = None
# YOUR CODE HERE
structuredEmailRDD = [Link](lambda row: extractEmailDetailsFromRawText(row['file,message']))
return structuredEmailRDD

16

structuredEmailRDD = createStructuredEmailRDD(emailRDD)
assert isinstance(structuredEmailRDD, RDD), "createStructuredEmailRDD() doesn't return an RDD"
assert isinstance([Link](), Row), "createStructuredEmailRDD() elements are not of type Row"

17

display([Link]())

Table

Date From To

1 null [] []

2 Thu, 1 Feb 2001 08:00:00 -0800 (PST) [] []

3 null ["[Link]@[Link]"] []

4 null [] ["[Link]@[Link]","","[Link]@[Link]","[Link]@enron.c…

[Link] (1).html 7/22

10:23 18/10/24 midterm1 - Databricks

5 null [] []

6 null [] []

7 null [] []

8 null [] []

9 null [] []

10 null [] []

11 null [] []

12 null [] []

13 null [] []

14 null [] []

10,000+ rows | Truncated data due to row limit

18

#1.
def countNumberEmail(structuredEmailRDD: RDD[Row], k: int)->int:
'''
The function counts the number of emails with more than `k` email addresses in the 'To' field.

Args:
- structuredEmailRDD: RDD of Row objects, each containing an email's structured data.
- k: The threshold for the number of emails in the 'To' field.

Returns:
- int: The count of emails with more than `k` email addresses in the 'To' field.
'''
count = -1
# YOUR CODE HERE
count = [Link](lambda row: len([Link]) > k).count()
# raise NotImplementedError()
return count

19

countNumberEmail(structuredEmailRDD,40)

print(countNumberEmail(structuredEmailRDD,40))

[Link] (1).html 8/22

10:23 18/10/24 midterm1 - Databricks

20

# 1đ
def countUniqueEmailDomains(structuredEmailRDD: RDD[Row], k) -> int:
'''
This function counts the number of unique email domains in the "CC" field using `map` and `reduce`.

Args:
- structuredEmailRDD: An RDD containing Row objects, each representing an email's structured data.

Returns:
- dict: A dictionary showing the count of emails from each unique domain in the "CC" field.
Example:
If k = 3, the result might look like:
{
'[Link]': 16452,
'[Link]': 122,
'[Link]': 101
}
'''
# results = {}
# YOUR CODE HERE

# Extract domains from CC field, ensuring valid email addresses

domainCounts = (
structuredEmailRDD
.flatMap(lambda row: [[Link]('@')[1] for email in [Link] if email and '@' in email]) # Extract domains safely
.map(lambda domain: (domain, 1)) # Create (domain, 1) tuples
.reduceByKey(lambda a, b: a + b) # Sum counts for each domain
)

# Collect domains that have a count greater than or equal to k

results = [Link](lambda x: x[1] >= k).collectAsMap()

# raise NotImplementedError()
return results

21

countUniqueEmailDomains(structuredEmailRDD, 10)

[Link] (1).html 9/22

10:23 18/10/24 midterm1 - Databricks

'[Link]': 20,
'[Link]': 12,
'[Link]': 31,
'[Link]': 19,
'EnronXGate': 49,
'EES': 204,
'[Link]': 12,
'[Link]': 25,
'[Link]': 29,
'ENRON=20': 11,
'[Link]>': 15,
'[Link]': 9122,
'[Link]': 17,
'[Link]>': 21,
'[Link]>': 94,
'[Link]': 15}

22

23

# 0.5đ ;
def countEmailsBySubject(structuredEmailRDD: RDD[Row], subjectContent: str) -> int:
'''
The function count num emails the dataset to the subject have included "subjectContent", case insensitive

Args:
- structuredEmailRDD: RDD of Row objects containing an email's structured data.
- subSubject: The content of the subject to filter by.

Returns:
- numEmails: Subject include content have substring subjectContent
'''
numEmails = -1
# YOUR CODE HERE
numEmails = (
structuredEmailRDD
.filter(lambda row: [Link] is not None and [Link]() in [Link]()) # Case insensitive check
.count() # Count the number of matching emails
)
# raise NotImplementedError()
return numEmails

[Link] (1).html 10/22

10:23 18/10/24 midterm1 - Databricks

24

countEmailsBySubject(structuredEmailRDD, "bank")

Out[22]: 126

25

26

[Link] (1).html 11/22

10:23 18/10/24 midterm1 - Databricks

import re

# 1.5d
def getTopKFrequentWordsInContentBySender(structuredEmailRDD: RDD[Row], sender: str, k: int) -> dict:
'''
This function filtered emails in the dataset sent by a specific sender's email,
and then returns the top k most frequent words found in the content of those filtered emails.

Args:
- structuredEmailRDD: RDD of Row objects, where each Row represents an email with structured data, such as subject and conte
- sender: sender's email which needs to count.
- k: The number of most frequent words to return.

Returns:
- topKWordsDict: A dictionary containing the top k most frequent words found in the content of filtered emails.
The keys are the words, and the values are their frequencies, representing how often they appear in the con
the filtered emails.
example:
{'to': 12,
'the': 9,
'your': 8,
'a': 5,
'is': 4,
'survey': 4,
'and': 4,
'you': 4,
'of': 3,
'very': 3}
'''
topKWordsDict = {}
# YOUR CODE HERE

# Filter emails from the specified sender

filteredEmails = [Link](lambda row: [Link] and [Link]() in [Link][0].lower())

# Extract words from the content of filtered emails

wordCounts = (
filteredEmails
.flatMap(lambda row: [Link](r'\w+', [Link]())) # Normalize words to lowercase and split into words
.map(lambda word: (word, 1)) # Create a tuple (word, 1)
.reduceByKey(lambda a, b: a + b) # Sum counts for each word
)

# Get the top k words by frequency

topKWords = [Link](k, key=lambda x: -x[1]) # Sort by frequency (descending)

# Convert the result to a dictionary

topKWordsDict = dict(topKWords)

# raise NotImplementedError()

[Link] (1).html 12/22

10:23 18/10/24 midterm1 - Databricks

return topKWordsDict

27

getTopKFrequentWordsInContentBySender(structuredEmailRDD, "[Link]@[Link]", 10)

Out[24]: {}

28

29

# SQL Query
from [Link] import SQLContext
from [Link] import col, size, to_date, year, month, weekofyear
[Link]("[Link]", "LEGACY")
# Convert RDD to DataFrame
dfEmais = [Link]()
dfEmais = [Link]('DateTime', to_date(col('Date'), "EEE, d MMM yyyy HH:mm:ss Z"))
dfEmais = [Link]('Num_To', size(col('To')))
dfEmais = [Link]('Num_CC', size(col('CC')))

30

sqlContext = SQLContext(sc)
tableName = "Emails"
[Link](tableName)

/databricks/spark/python/pyspark/sql/[Link]: FutureWarning: Deprecated in 3.0.0. Use [Link]() i

nstead.
[Link](

31

[Link] (1).html 13/22

10:23 18/10/24 midterm1 - Databricks

# 0.5đ
def getTopKRowsBySQL(sqlContext: SQLContext, tableName: str, k: int) -> DataFrame:
'''
This function queries the first k rows from a given table using SQLContext.

Args:
- sqlContext: An SQLContext object that provides the environment to run SQL queries on structured data.
- tableName: The name of the table from which the rows will be selected.
- k: The number of rows to return. It must be a positive integer.

Returns:
- result: A DataFrame containing the first k rows of data from the specified table.
If the table contains fewer than k rows, the DataFrame will contain all available rows.
'''
# YOUR CODE HERE
# raise NotImplementedError()
result = [Link](f"SELECT * FROM {tableName} LIMIT {k}")
return result

def getTopKRowsByDFOperations(dataFrame: DataFrame, k: int) -> DataFrame:

'''
This function queries the first k rows from a given table using SQLContext.

Args:
- dataFrame: An DataFrame data
- tableName: The name of the table from which the rows will be selected.
- k: The number of rows to return. It must be a positive integer.

Returns:
- resultDF: A DataFrame containing the first k rows of data from the specified table.
If the table contains fewer than k rows, the DataFrame will contain all available rows.
'''
# YOUR CODE HERE
# raise NotImplementedError()
result = [Link](k)

return result

32

getTopKRowsBySQL(sqlContext,tableName,1).show()

+----+----+---+-------+---+-------+--------+------+------+
|Date|From| To|Subject| CC|Content|DateTime|Num_To|Num_CC|
+----+----+---+-------+---+-------+--------+------+------+
|null| []| []| null| []| | null| 0| 0|
+----+----+---+-------+---+-------+--------+------+------+

[Link] (1).html 14/22

10:23 18/10/24 midterm1 - Databricks

33

getTopKRowsByDFOperations(dfEmais,1).show()

+--------------------+
| file,message|
+--------------------+
|jones-t/sent/5567...|
+--------------------+

34

[Link] (1).html 15/22

10:23 18/10/24 midterm1 - Databricks

# 1đ, 6.5đ
def countEmailsWithToGreaterThanKBySQL(sqlContext: SQLContext, tableName: str, k: int) -> int:
'''
This function uses SQL to count the number of emails with more than k email addresses in the To field.

Args:
- sqlContext: An SQLContext object that provides the environment to run SQL queries on structured data.
- tableName: The name of the table that contains the email data.
- k: The threshold number of email addresses in the To field.

Returns:
- count: An integer representing the number of emails where the To field has more than k email addresses.
'''
# YOUR CODE HERE

# Do cột To là kiểu array nên chỉ dùng size(To) > {k} còn không thì size(split(To, ',')) > {k}
result = [Link](f"SELECT count(*) AS count_email FROM {tableName} WHERE size(To) > {k}")
count = [Link]()[0]['count_email'] # Lấy giá trị đếm

# raise NotImplementedError()
# # Kiểm tra xem k có phải là số nguyên dương không
# if k <= 0:
# raise ValueError("k must be a positive integer")

# # Thực hiện truy vấn SQL để lấy k hàng đầu tiên

# query = f"SELECT * FROM {tableName} LIMIT {k}"
# count = [Link](query)

return count

def countEmailsWithToGreaterThanKByDFOperations(dataFrame: DataFrame, k: int) -> int:

'''
This function counts the number of emails with more than k email addresses in the To field using DataFrame operations.

Args:
- dataFrame: A DataFrame containing the email data.
- k: The threshold number of email addresses in the To field.

Returns:
- count: An integer representing the number of emails where the To field has more than k email addresses.
'''
# YOUR CODE HERE

# Sử dụng các phép toán trên DataFrame để đếm số email có nhiều hơn k địa chỉ trong trường 'To'
result = [Link]([Link]()) \
.filter(size(split([Link], ',')) > k) \
.count() # Đếm số lượng email thỏa mãn điều kiện
# raise NotImplementedError()
# Kiểm tra trường 'To' không bị null
result = [Link]([Link]()) \

[Link] (1).html 16/22

10:23 18/10/24 midterm1 - Databricks

.filter([Link] != '') \
.filter([Link](0) != '') \
.filter([Link](0).isNotNull()) \
.selectExpr("size(split(To, ',')) as emailCount") \
.filter("emailCount > {}".format(k)) \
.count() # Đếm số lượng email thỏa mãn điều kiện

return result

35

[Link](f"DESCRIBE {tableName}").show()

+--------+-------------+-------+
|col_name| data_type|comment|
+--------+-------------+-------+
| Date| string| null|
| From|array<string>| null|
| To|array<string>| null|
| Subject| string| null|
| CC|array<string>| null|
| Content| string| null|
|DateTime| date| null|
| Num_To| int| null|
| Num_CC| int| null|
+--------+-------------+-------+

36

countEmailsWithToGreaterThanKBySQL(sqlContext,tableName,50)

Out[32]: 0

37

1 # YOUR CODE HERE

2 raise NotImplementedError()

NotImplementedError:

[Link] (1).html 17/22

10:23 18/10/24 midterm1 - Databricks

38

countEmailsWithMoreToThanCCInJanuaryBySQL(sqlContext, tableName)

39

countEmailsWithMoreToThanCCInJanuaryByDFOperations(dfEmais)

Command skipped

40

41

42

[Link] (1).html 18/22

10:23 18/10/24 midterm1 - Databricks

# 1đ, 8.5
def countEmailsInMonthBySQL(sqlContext: SQLContext, tableName: str, monthValue: int) -> int:
'''
This function calculates the number of emails sent in a given year by performing a group by operation on the DateTime column

Args:
- sqlContext: An SQLContext object that provides the environment to run SQL queries on structured data.
- tableName: The name of the table that contains the email data.
- monthValue: The specific month to filter the emails by (1-12).

Returns:
- email_count: An integer representing the number of emails sent in the given date.
'''
# YOUR CODE HERE
raise NotImplementedError()
return emailCount

def countEmailsInMonthByDFOperations(dataFrame: DataFrame, monthValue: int) -> int:

'''
This function calculates the number of emails sent in a given month by performing a filter operation on the DateTime column
DataFrame operations.

Args:
- dataFrame: A DataFrame containing the email data.
- monthValue: The specific date to filter the emails by.

Returns:
- email_count: An integer representing the number of emails sent in the given monthValue.
'''
assert isinstance(dataFrame, DataFrame), f"dataFrame is not the correct data type, expected DataFrame but got {type(dataFram
assert isinstance(monthValue, int) and monthValue > 0 and monthValue <= 12, f"date must be a positive integer, but got {mont
# YOUR CODE HERE
raise NotImplementedError()
return result

Command skipped

43

countEmailsInMonthBySQL(sqlContext,tableName,5)

Command skipped

44

countEmailsInMonthByDFOperations(dfEmais,5)

[Link] (1).html 19/22

10:23 18/10/24 midterm1 - Databricks

Command skipped

45

[Link] (1).html 20/22

10:23 18/10/24 midterm1 - Databricks

# 1đ
def countEmailsPerWeekBySQL(sqlContext: SQLContext, tableName: str) -> DataFrame:
'''
This function calculates the number of emails sent per Week by performing a group by operation on the DateTime column
and sorts the result in descending order by the number of emails.

Args:
- sqlContext: An SQLContext object that provides the environment to run SQL queries on structured data.
- tableName: The name of the table that contains the email data.

Returns:
- resultDF: A DataFrame containing the number of emails sent per Week, sorted in descending order by num week.
With schema:
root
|-- week: integer (nullable = true)
|-- count: long (nullable = false)

'''
assert isinstance(sqlContext, SQLContext), f"sqlContext is not the correct data type, expected SQLContext but got {type(sqlC
assert isinstance(tableName, str), f"tableName is not the correct data type, expected string but got {type(tableName)}"
# YOUR CODE HERE

result = [Link](f"SELECT weekofyear(DateTime) as week, count(*) as count FROM {tableName} GROUP BY week ORDER BY cou

# raise NotImplementedError()
return result

def countEmailsPerWeekByDFOperations(dataFrame: DataFrame) -> DataFrame:

'''
This function calculates the number of emails sent per Week by performing a group by operation on the DateTime column
and sorts the result in descending order by the number of emails.

Args:
- dataFrame: A DataFrame containing the email data.

Returns:
- resultDF: A DataFrame containing the number of emails sent per Week, sorted in descending order by num week.
With schema:
root
|-- week: integer (nullable = true)
|-- count: long (nullable = false)
'''
assert isinstance(dataFrame, DataFrame), f"dataFrame is not the correct data type, expected DataFrame but got {type(dataFram
# YOUR CODE HERE

# Tính toán số tuần từ cột DateTime và nhóm theo tuần

result = ([Link]("week", weekofyear([Link]))
.groupBy("week")
.count()

[Link] (1).html 21/22

10:23 18/10/24 midterm1 - Databricks

.orderBy("count", ascending=False))

# raise NotImplementedError()
return result

Command skipped

46

countEmailsPerWeekBySQL(sqlContext,tableName).show()

Command skipped

47

Command skipped

[Link] (1).html 22/22