我需要获得一个增量表的分区列,但是a的返回结果DESCRIBE delta.`my_table`
在数据块上和PyCharm本地返回不同的结果。
最小的例子:
from pyspark.sql.types import StructType, StructField, StringType, IntegerType
delta_table_path = "c:/temp_delta_table"
partition_column = ["rs_nr"]
schema = StructType([
StructField("rs_nr", StringType(), False),
StructField("event_category", StringType(), True),
StructField("event_counter", IntegerType(), True)])
data = [{'rs_nr': '001', 'event_category': 'event_01', 'event_counter': 1},
{'rs_nr': '002', 'event_category': 'event_02', 'event_counter': 2},
{'rs_nr': '003', 'event_category': 'event_03', 'event_counter': 3},
{'rs_nr': '004', 'event_category': 'event_04', 'event_counter': 4}]
sdf = spark.createDataFrame(data=data, schema=schema)
sdf.write.format("delta").mode("overwrite").partitionBy(partition_column).save(delta_table_path)
df_descr = spark.sql(f"DESCRIBE delta.`{delta_table_path}`")
df_descr.toPandas()
在数据块上显示分区列:
col_name data_type comment
0 rs_nr string None
1 event_category string None
2 event_counter int None
3 # Partition Information
4 # col_name data_type comment
5 rs_nr string None
但是当在PyCharm本地运行时,我得到以下不同的输出:
col_name data_type comment
0 rs_nr string
1 event_category string
2 event_counter int
3
4 # Partitioning
5 Part 0 rs_nr
解析两种类型的返回值对我来说似乎很难看,所以这是这样返回的原因吗?
设置:
在Pycharm:
- pyspark = 3.2.3
- delta-spark = 2.0.0在砖:
- DBR 11.3 LTS
- Spark = 3.3.0(我只是注意到这是不同的,我将在此期间测试3.3.0是否在本地工作) Scala = 2.12
在PyCharm中,我使用以下命令创建连接:
def get_spark():
spark = SparkSession.builder.appName('schema_checker')
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
.config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
.config("spark.jars.packages", "io.delta:delta-core_2.12:2.0.0")
.config("spark.sql.catalogImplementation", "in-memory")
.getOrCreate()
return spark
如果您正在使用Python,那么与其执行更难解析的SQL命令,不如使用Python API。DeltaTable
实例有一个detail
函数,它返回一个包含表(doc)详细信息的数据框,这个数据框有partitionColumns
列,它是一个带有分区列名称的字符串数组。所以你可以输入:
from delta.tables import *
detailDF = DeltaTable.forPath(spark, delta_table_path).detail()
partitions = detailDF.select("partitionColumns").collect()[0][0]