更新dazfuller建議,但包括代碼級別的分區,
當然如果你有更深層次的分區,那麼你將不得不做出一個函數和遞歸調用最後包含鑲花的文件目錄。
鑲花的模式除了分區,您必須添加,
我選擇分區的數據類型是字符串,不確定你可以推斷出更多,老實說不確定重要的火花優化。
這是代碼. .
#創建一個工作目錄data_path = " / tmp / directory_example”#刪除內容#如果運行不止一次或改變dbutils.fs有用。rm (data_path,真)dbutils.fs.mkdirs (data_path)
#創建一個數據庫database_name = " tomh_directory_test " #清楚數據庫是否已經存在火花。sql (f“刪除數據庫,如果存在{database_name}級聯”)火花。sql (f“創建數據庫{database_name}”)火花。sql (f“使用{database_name}”)
創建一個dataframe編寫示例內容。
我在兩個不同的分區方式。
你可能想要測試如果你有更深層次的嵌套分區為例。
模式= " int ID、名稱字符串,字符串,字符串“data =[[1,“查爾斯”、“培訓”、“2020”),(2)“伊萬”、“銷售”、“2021”],[3,“多蘿西”、“培訓”、“2020”],[4,“傑森”、“銷售”、“2021”]]df =火花。createDataFrame(數據、模式).coalesce(1)顯示(df)
寫分區的數據
table_name = " / table1 " partition_path = f“{data_path} {table_name}”打印(partition_path) df.write.format .partitionBy(“鋪”)(“部門”).save (partition_path)
寫數據分區
table_name = " /表“partition_path = f”{data_path} {table_name}”打印(partition_path) df.write.format .partitionBy(“鋪”)(“年”).save (partition_path)
當前的狀態
我們有一個目錄有兩個子目錄,每個分區的不同的列
下一個步驟
編寫一些代碼來生成類似的語句
創建表名(int ID、名稱字符串,字符串,字符串)使用拚花分區(部門)位置“/ tmp / directory_example table1”;
這將是你如果你想添加一個“如果不存在條款”create table語句。隻有創建新目錄表。
表名
我使用的目錄表的名字
分區
如果拚花的文件有子目錄,這些將是我們分區,我隻是轉深一層,您可能需要遞歸直到你點擊數據文件如果一些表有多個分區
表的位置
這將是一樣的目錄表的一個名字
表模式
表模式將一個模式的組合生成通過閱讀一個分區的文件夾和添加分區列。
注意如果空分區存在您必須捕獲和閱讀另一個分區
創建表
我們將使用外部表,通過定義外部表的位置。
MSCK
代碼運行MSCK修理表更新metastore檢查分區。
basedir = dbutils.fs.ls (data_path) #打印(basedir) # partition_list = [] x basedir:如果x [1] .endswith(“/”): #有子目錄#使用目錄名稱表名# basedir行對象的列表是#項指數1 x[1]是文件名table_name = x[1](: 1) #刪除最後的性格/打印(f表名稱:{table_name}) #得到相同表位置的文件對象列表# x[0]是完整路徑table_location = x[0]打印(f表位置:{table_location}) #分區的列列表的行對象從dbutils #。fs上麵的子目錄#邏輯一樣,獲得的名稱的目錄名和解析分區列path_to_one_subdirectory = dbutils.fs.ls (x[0])[0][0] #這將是一個目錄,鑲花的文件你需要一個遞歸調用如果使用子分區partition_column = (dbutils.fs.ls (x [0]) [0] [1] .split(" = ")[0])打印(f“分區列是:{partition_column}”) #得到基地模式#從數據目錄中,創建一個dataframe df = spark.read.format持有鑲花的文件(“鋪”).load (path_to_one_subdirectory) base_schema = df.schema.simpleString () [7: 1]。替換(“:”、“”)#打印(myschema [7: 1]。替換(“:””、“)#打印(base_schema) #添加分區列full_schema = f“{base_schema}, {partition_column}字符串”#打印(“# # # # # #”)打印(f“完整的模式是:{full_schema}”) #打印(# # # # # # #)#構建create table語句create_table_statement = f“創建表{table_name} ({full_schema})使用鑲花({partition_column})位置分區的{table_location}”打印(f“創建table語句:{create_table_statement}”) #創建表spark.sql (create_table_statement)火花。sql (f“msck修理表{table_name}”)打印(“——”)
結束的時候,你應該在當前數據庫表1和表二。
從表二sql SELECT * %;
從表1 sql SELECT * %;
清理這段代碼如果你跑這一切作為一個實驗。
#清理dbutils.fs。rm (data_path,真的)
#清理火花。sql(“刪除數據庫tomh_directory_test級聯”)
我附上了這個演示的筆記本。
