PythonSQL应用随笔4——PySpark创建SQL临时表

零、前言

Python中直接跑SQL，可以很好的解决数据导过来导过去的问题，本文方法主要针对大运算量时，如何更好地让Python和SQL打好配合。

工具：Zeppelin
语法：PySpark（Apache Spark的Python API）、SparkSQL
数据库类型：Hive

一、相关方法

.createOrReplaceTempView()
在PySpark中，createOrReplaceTempView是一个用于DataFrame的方法，它允许你将DataFrame的内容注册为一个临时的SQL视图，这样就可以在Spark SQL查询中引用这个视图，就像正常查询常规数仓表一样。
.toPandas()
最终取数结果，以DataFrame形式输出。

二、实例

Zeppelin中编辑器与Jupyter Notebook类似，以代码块形式呈现，只是需要提前指定好代码块的语言，如：%pyspark；
日常工作中，库存数据是常见的大数据量取数场景，下述代码以取 sku每天的库存为例展开。

%pyspark
# 工具包及基础配置（视具体情况进行配置，非本文重点，可略过）
import pandas as pd
from pyspark import SparkConf
from pyspark import SparkContext
from pyspark.sql.types import *
from pyspark.sql import SparkSession
from pyspark.sql import SQLContextspark_conf = SparkConf()
spark_conf.setMaster("local[*]")
spark_conf.setAppName("Test")
spark_conf.set("zeppelin.spark.sql.stacktrace", "true")
spark_conf.set('hive.exec.dynamic.partition.mode', 'nonstrict')
spark_conf.set("spark.sql.execution.arrow.enabled", "true")
spark_conf.set("spark.sql.execution.arrow.fallback.enabled", "true")
spark = SparkSession.builder.config(conf=spark_conf).config("zeppelin.spark.sql.stacktrace", "true").enableHiveSupport().getOrCreate()

%pyspark
# 配置取数参数（省事小技巧，避免重复编码，根据实际情况可配置多个参数）
## 开始、结束日期、品牌、……
start_date = '2024-01-01'
end_date = '2024-01-31'
brand = 'brand01'# sql1：日期维表
tmp_dim_date = '''select date_stringfrom edw.dim_datewhere 1=1and date_string >= '{start_date}'and date_string <= '{end_date}''''.format(start_date=start_date, end_date=end_date)
tmp_dim_date = spark.sql(tmp_dim_date).createOrReplaceTempView('tmp_dim_date') # 创建日期临时表：tmp_dim_date# sql2：商品维表
tmp_dim_sku = '''select brand_name,sku_skfrom edw.dim_skuwhere 1=1and brand_name = '{brand}'group by 1, 2'''.format(brand=brand)
tmp_dim_sku = spark.sql(tmp_dim_sku).createOrReplaceTempView('tmp_dim_sku') # 创建sku临时表：tmp_dim_sku# 最终sql：sku每天的库存
sku_stock = '''select tb0.date_string,tb1.sku_sk,sum(coalesce(tb1.stock_qty, 0)) stock_qty -- 库存量from tmp_dim_date tb0 -- 日期临时表left join edw.stock_zipper tb1 -- 库存拉链表on tb1.date_begin <= tb0.date_string -- 开链时间and tb1.date_end > tb0.date_string -- 闭链时间inner join tmp_dim_sku tb2 -- sku临时表on tb1.sku_sk = tb2.sku_skgroup by 1, 2'''
df_sku_stock = spark.sql(tmp_stock_zipper).toPandas()# 删除临时视图（在不需要时及时做垃圾回收，减少资源占用）
spark.catalog.dropTempView("tmp_dim_stockorg")
spark.catalog.dropTempView("tmp_dim_sku")

至此，sku天维度库存数据已取出，实际应用常见可能比本案例复杂许多，故临时表的方法才更重要，一方面能理清楚取数代码的结构，一方面也提高代码性能。