Scribd
Upload
10 views14 pages

Bookstore Dataset Copy and Structure

Uploaded by

jimex63897
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as PDF, TXT or read online on Scribd
10 views14 pages

Bookstore Dataset Copy and Structure

Uploaded by

jimex63897
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as PDF, TXT or read online on Scribd

([Link]

com)
MOCK (Python)

Import notebook
1

%run "/Workspace/Users/fy25shiftleftob29auguser65@[Link]/hon mock/Includes/Copy-Datasets"

Copying books-cdc/ ...


Copying books-csv-new/ ...
Copying books-csv/ ...
Copying books-streaming/ ...
Copying customers-json-new/ ...
Copying customers-json/ ...
Copying orders-json-raw/ ...
Copying orders-json-streaming/ ...
Copying orders-new/ ...
Copying orders-raw/ ...
Copying orders-streaming/ ...
Copying orders/ ...

%fs ls 'dbfs:/FileStore/demo-datasets/bookstore'

Table

path name size modificationTime

1 dbfs:/FileStore/demo-datasets/bookstore/books-cdc/ books-cdc/ 0 1758554045000

2 dbfs:/FileStore/demo-datasets/bookstore/books-csv/ books-csv/ 0 1758554049000

3 dbfs:/FileStore/demo-datasets/bookstore/books-csv-new/ books-csv-new/ 0 1758554047000

4 dbfs:/FileStore/demo-datasets/bookstore/books-streaming/ books-streaming/ 0 1758554052000

5 dbfs:/FileStore/demo-datasets/bookstore/customers-json/ customers-json/ 0 1758554060000

6 dbfs:/FileStore/demo-datasets/bookstore/customers-json-new/ customers-json-new/ 0 1758554058000

7 dbfs:/FileStore/demo-datasets/bookstore/orders/ orders/ 0 1758554082000

8 dbfs:/FileStore/demo-datasets/bookstore/orders-json-raw/ orders-json-raw/ 0 1758554064000

9 dbfs:/FileStore/demo-datasets/bookstore/orders-json-streaming/ orders-json-streaming/ 0 1758554066000

10 dbfs:/FileStore/demo-datasets/bookstore/orders-new/ orders-new/ 0 1758554072000

11 dbfs:/FileStore/demo-datasets/bookstore/orders-raw/ orders-raw/ 0 1758554074000

12 dbfs:/FileStore/demo-datasets/bookstore/orders-streaming/ orders-streaming/ 0 1758554075000

12 rows

; 3

%fs ls 'dbfs:/FileStore/demo-datasets/bookstore/books-csv/'

Table
%fs head dbfs:/FileStore/demo-datasets/bookstore/books-csv/export_001.csv

book_id;title;author;category;price
B10;Beginning Database Design Solutions;Rod Stephens;Computer Science;44
B11;Business Intelligence for Dummies;Swain Scheps;Computer Science;38
B12;Big Data in Practice;Bernard Marr;Computer Science;30

%fs ls dbfs:/FileStore/demo-datasets/bookstore/customers-json/

Table

%fs head dbfs:/FileStore/demo-datasets/bookstore/customers-json/export_001.json

[Truncated to first 65536 bytes]


{"customer_id":"C00001","email":"dabby2y@[Link]","profile":"{\"first_name\":\"Dniren\",\"last_name\":\"A
bby\",\"gender\":\"Female\",\"address\":{\"street\":\"768 Mesta Terrace\",\"city\":\"Annecy\",\"country\":\"Fr
ance\"}}","updated":"2021-12-14T23:15:43.375Z"}
{"customer_id":"C00002","email":"eabbysc1@[Link]","profile":"{\"first_name\":\"Etti\",\"last_name\":\"Abby
s\",\"gender\":\"Female\",\"address\":{\"street\":\"1748 Vidon Plaza\",\"city\":\"Varge Mondar\",\"country\":
\"Portugal\"}}","updated":"2021-12-14T23:15:43.375Z"}
{"customer_id":"C00003","email":"rabelovd1@[Link]","profile":"{\"first_name\":\"Ronnie\",\"last_name
\":\"Abelov\",\"gender\":\"Male\",\"address\":{\"street\":\"363 Randy Park\",\"city\":\"San Celestio\",\"count
ry\":\"Philippines\"}}","updated":"2021-12-14T23:15:43.375Z"}
{"customer_id":"C00004","email":"rabels9g@[Link]","profile":"{\"first_name\":\"Ray\",\"last_name\":\"Abel
s\",\"gender\":\"Female\",\"address\":{\"street\":\"613 Lyons Way\",\"city\":\"Oudtshoorn\",\"country\":\"Sout
h Africa\"}}","updated":"2021-12-14T23:15:43.375Z"}
{"customer_id":"C00005","email":"sabendrothin@[Link]","profile":"{\"first_name\":\"Shanon\",\"las
t_name\":\"Abendroth\",\"gender\":\"Female\",\"address\":{\"street\":\"30292 Manufacturers Junction\",\"city
\":\"Ani-e\",\"country\":\"Philippines\"}}","updated":"2021-12-14T23:15:43.375Z"}
{"customer_id":"C00006","profile":"{\"first_name\":\"Norman\",\"last_name\":\"Abernethy\",\"gender\":\"Male\",
\"address\":{\"street\":\"9292 Oxford Center\",\"city\":\"Gibara\",\"country\":\"Cuba\"}}","updated":"2021-12-
14T23:15:43.375Z"}
{"customer_id":"C00007","email":"sabrahmson3h@[Link]","profile":"{\"first_name\":\"Skell\",\"last_name
\":\"Abrahmson\",\"gender\":\"Male\",\"address\":{\"street\":\"90941 Hallows Park\",\"city\":\"Huarong Chenggu

%fs ls 'dbfs:/FileStore/demo-datasets/bookstore/orders-raw/'
Table

df_orders = [Link]("dbfs:/FileStore/demo-datasets/bookstore/orders-raw/*.parquet")
display(df_orders.show(20))
df_orders.createOrReplaceTempView("orders_vw")

  df_orders: [Link] = [order_id: string, order_timestamp: long ... 4 more fields]


|000000000006343| 1657531717| C00654| 1| 28|[{B02, 1, 28}]|
|000000000006344| 1657531717| C00654| 1| 28|[{B02, 1, 28}]|
|000000000006345| 1657543676| C00762| 1| 49|[{B01, 1, 49}]|
|000000000006346| 1657543676| C00762| 1| 49|[{B01, 1, 49}]|
|000000000006347| 1657546079| C01014| 1| 28|[{B02, 1, 28}]|
|000000000006348| 1657546658| C00633| 1| 24|[{B09, 1, 24}]|
|000000000006349| 1657546658| C00633| 1| 24|[{B09, 1, 24}]|
|000000000006350| 1657547177| C00638| 1| 35|[{B03, 1, 35}]|
|000000000006351| 1657547177| C00638| 1| 35|[{B03, 1, 35}]|
|000000000006352| 1657547578| C00875| 1| 24|[{B09, 1, 24}]|
|000000000006353| 1657554241| C01130| 1| 41|[{B08, 1, 41}]|
|000000000006354| 1657554482| C00753| 1| 33|[{B07, 1, 33}]|
|000000000006355| 1657554482| C00753| 1| 33|[{B07, 1, 33}]|
|000000000006356| 1657555151| C01177| 1| 20|[{B04, 1, 20}]|
|000000000006357| 1657555693| C00639| 1| 33|[{B07, 1, 33}]|
|000000000006358| 1657555693| C00639| 1| 33|[{B07, 1, 33}]|
|000000000006359| 1657557010| C00942| 1| 24|[{B09, 1, 24}]|
|000000000006360| 1657562811| C00760| 1| 47|[{B05, 1, 47}]|
+---------------+---------------+-----------+--------+-----+--------------+
only showing top 20 rows

%python

distinct=(df_orders.select("customer_id").where ("customer_id is not null").[Link](lambda


x:x[0]).distinct().count())
print(distinct)

700

%python
#Find the price of the book B10 dataset using Spark RDD.
book_rdd=[Link]("header","true").option("inferSchema","true").option('delimiter',';').csv("d
bfs:/FileStore/demo-datasets/bookstore/books-csv").rdd

books10=book_rdd.filter(lambda x:x[0]=='B10').map(lambda x:x[4]).collect()


print(books10)

[44]

#Write a Spark SQL query to calculate the average price of books per customer.
df_books=[Link]("header","True").option("delimiter",";").option("inferSchema","True").csv("d
bfs:/FileStore/demo-datasets/bookstore/books-csv/*.csv")

  df_books: [Link]
book_id: string
title: string
author: string
category: string
price: integer

df_books.createOrReplaceTempView("books_Vw")

df_customers=[Link]("dbfs:/FileStore/demo-datasets/bookstore/customers-json/")
df_customers.createOrReplaceTempView("customers_vw")

  df_customers: [Link] = [customer_id: string, email: string ... 2 more fields]

%sql
--DELTA TABLE OF CUSTOMERS
create or replace table customers as select * from `customers_vw`;

  _sqldf: [Link] = [num_affected_rows: long, num_inserted_rows: long]

Table

This result is stored as _sqldf and can be used in other Python cells.

%sql
-------DELTA TABLE OF ORDERS
create or replace table orders as select * from orders_vw;

  _sqldf: [Link] = [num_affected_rows: long, num_inserted_rows: long]


Table

This result is stored as _sqldf and can be used in other Python cells.

%sql
-------DELTA TABLE OF BOOKS
create or replace table books
as select * from books_Vw;

  _sqldf: [Link] = [num_affected_rows: long, num_inserted_rows: long]

Table

This result is stored as _sqldf and can be used in other Python cells.

%sql

--#Write a Spark SQL query to calculate the average price of books per customer.
with ordersdetails as (select customer_id, explode(books) as book_item from orders)
select o.customer_id,avg([Link]) as avg_price from ordersdetails o join books b on
b.book_id=o.book_item.book_id
group by o.customer_id order by o.customer_id;

  _sqldf: [Link] = [customer_id: string, avg_price: double]


Table

This result is stored as _sqldf and can be used in other Python cells.

from [Link] import from_json, col


from [Link] import StructType, StructField, StringType

# Define the schema for the nested 'profile' JSON string


profile_schema = StructType([
StructField("first_name", StringType(), True),
StructField("last_name", StringType(), True),
StructField("gender", StringType(), True),
StructField("address", StructType([
StructField("street", StringType(), True),
StructField("city", StringType(), True),
StructField("country", StringType(), True)
]), True)
])

customers_json_path = "dbfs:/FileStore/demo-datasets/bookstore/customers-json/*.json"
df_customers_raw = [Link]("json").load(customers_json_path)

# Parse the profile string into a struct and create a clean table
df_customers_parsed = (df_customers_raw
.withColumn("profile_struct", from_json(col("profile"), profile_schema))
.select("customer_id", "email", col("profile_struct").alias("profile"), "updated")
)

# Overwrite the existing table with the correctly-typed data


df_customers_parsed.[Link]("delta") \
.option("overwriteSchema", "true") \
.mode("overwrite") \
.saveAsTable("customers")

print("Table 'customers' has been recreated with a structured 'profile' column.")

  df_customers_parsed: [Link] = [customer_id: string, email: string ... 2 more fields]


  df_customers_raw: [Link] = [customer_id: string, email: string ... 2 more fields]
Table 'customers' has been recreated with a structured 'profile' column.
%sql
--2. Write a Spark SQL query to find customers with missing or incomplete information
SELECT
customer_id,
email,
profile.first_name,
profile.last_name,
[Link]
FROM
customers
WHERE
email IS NULL
OR profile.first_name IS NULL
OR profile.last_name IS NULL
OR [Link] IS NULL
OR [Link] IS NULL
OR [Link] IS NULL;

  _sqldf: [Link] = [customer_id: string, email: string ... 3 more fields]

Table

This result is stored as _sqldf and can be used in other Python cells.

%sql
--2. find the Total price for book B11 and how many customers have ordered it
with ordersdetails as (select customer_id, explode(books) as book_item from orders)
select b.book_id,sum([Link]) as totalPrice,count(o.customer_id) as TotalCustomers from books b join
ordersdetails o on b.book_id=o.book_item.book_id where b.book_id='B11'
group by b.book_id;

  _sqldf: [Link] = [book_id: string, totalPrice: long ... 1 more field]


Table

This result is stored as _sqldf and can be used in other Python cells.

%sql
--3. Demonstrate how to use the VACUUM command with a retention period to retain only the most
recent versions of files for the "customers" Delta table.
vacuum customers retain 168 hours;

  _sqldf: [Link] = [path: string]

Table

This result is stored as _sqldf and can be used in other Python cells.

%sql
describe history customers;

  _sqldf: [Link] = [version: long, timestamp: timestamp ... 13 more fields]

Table

This result is stored as _sqldf and can be used in other Python cells.

[Link]('dbfs:/FileStore/demo-datasets/bookstore/orders-raw/'
,'dbfs:/FileStore/tables/orders/streaming',True)

True

b. Create Streaming table orders_streaming and upload the data from the directory

df_orders.printSchema()

root
|-- order_id: string (nullable = true)
|-- order_timestamp: long (nullable = true)
|-- customer_id: string (nullable = true)
|-- quantity: long (nullable = true)
|-- total: integer (nullable = true)
|-- books: array (nullable = true)
| |-- element: struct (containsNull = true)
| | |-- book_id: string (nullable = true)
| | |-- quantity: integer (nullable = true)
| | |-- subtotal: long (nullable = true)

from [Link] import *


books_schema=(StructType([StructField('book_id', StringType(),True),StructField('quantity',
IntegerType(),True),StructField('subtotal', LongType(),True)]))

schema_orders=(StructType
([
StructField ('order_id',StringType(),True),
StructField('order_timestamp', LongType(),True),
StructField('customer_id', StringType(),True),
StructField('quantity',LongType(),True),
StructField('total',IntegerType(),True),
StructField('books',ArrayType(books_schema),True)
]))

#Streaming
ordder_streaming=[Link](schema_orders).option("header","true").parquet("dbfs:/FileStor
e/tables/orders/streaming")

  ordder_streaming: [Link] = [order_id: string, order_timestamp: long ... 4 more fields]

ordder_streaming.isStreaming

True

#Autoloader
df_stream=
([Link]("cloudFiles").option("[Link]","parquet").option("[Link]
ocation","dbfs:/FileStore/tables/orders/streaming/schemaas/").load("dbfs:/FileStore/tables/orders/strea
ming"))
  df_stream: [Link]
order_id: string
order_timestamp: long
customer_id: string
quantity: long
total: integer
 books: array
 element: struct
book_id: string
quantity: integer
subtotal: long
_rescued_data: string

df_stream.createOrReplaceTempView("orders_temp_vw")

streaming_order.[Link]("delta").option("checkpointLocation","dbfs:/FileStore/tables/orders/
checkpoint").table("orders_parquet")

8cfc7e32-4426-4431-926c-43ae5dd9664f Last updated: 2 hours ago

<[Link] at 0x7fc98f541420>

streaming_order=[Link]("select * from orders_temp_vw")

  streaming_order: [Link] = [order_id: string, order_timestamp: long ... 5 more fields]

%sql
DROP TABLE IF EXISTS orders

OK

%fs rm -r "dbfs:/FileStore/tables/orders/streaming "

res17: Boolean = false

%sql
DROP TABLE IF EXISTS orders_stream

OK

%sql
show tables;

  _sqldf: [Link] = [database: string, tableName: string ... 1 more field]


Table

This result is stored as _sqldf and can be used in other Python cells.

%sql
--e. Write an SQL query to retrieve total quantity of each item from the orders_streaming table
with ordersdetails as (select Quantity, explode(books) as book_item from orders_parquet)
select sum(Quantity) as Total_Quantity,book_item.book_id from ordersdetails group by book_item.book_id
order by book_item.book_id;

  _sqldf: [Link] = [Total_Quantity: long, book_id: string]

Table

This result is stored as _sqldf and can be used in other Python cells.

[Link]('dbfs:/FileStore/demo-datasets/bookstore/orders-
new/','dbfs:/FileStore/tables/orders/streaming',True)

True

%sql
--e. Write an SQL query to retrieve total quantity of each item from the orders_streaming table
with ordersdetails as (select Quantity, explode(books) as book_item from orders_parquet)
select sum(Quantity) as Total_Quantity,book_item.book_id from ordersdetails group by book_item.book_id
order by book_item.book_id;

  _sqldf: [Link] = [Total_Quantity: long, book_id: string]

Table

This result is stored as _sqldf and can be used in other Python cells.

%sql
--Write an SQL query to retrieve total price of each customer in real time whenever new data is being
ingested.
with ordersdetails as ( select Quantity,customer_id,explode(books) as book_item from orders_parquet)
select c.customer_id,sum([Link]*[Link]) as Total_price from books b join ordersdetails c on
b.book_id=c.book_item.book_id group by c.customer_id order by c.customer_id;

  _sqldf: [Link] = [customer_id: string, Total_price: long]


Table

This result is stored as _sqldf and can be used in other Python cells.

  _sqldf: [Link] = [path: string, metrics: struct]

Table

This result is stored as _sqldf and can be used in other Python cells.

You might also like