python spark-sql-temp-table.py

Posted 2021-05-09

from pyspark.sql import Row
ordersRDD = sc.textFile("/public/retail_db/orders")
ordersDF = ordersRDD.\
map(lambda o: Row(order_id=int(o.split(",")[0]), order_date=o.split(",")[1], order_customer_id=int(o.split(",")[2]), order_status=o.split(",")[3])).toDF()
ordersDF.registerTempTable("ordersDF_table")
sqlContext.sql("select order_status, count(1) from ordersDF_table group by order_status").show()

sqlContext.sql("use dgadiraju_retail_db_txt")
from pyspark.sql import Row
productsRaw = open("/data/retail_db/products/part-00000").read().splitlines()
productsRDD = sc.parallelize(productsRaw)
productsDF = productsRDD.\
map(lambda p: Row(product_id=int(p.split(",")[0]), product_name=p.split(",")[2])).\
toDF()
productsDF.registerTempTable("products")

sqlContext.sql("select * from products").show()
sqlContext.sql("select * from orders").show()
sqlContext.sql("select * from order_items").show()