Python Version 3.7.5Spark Version 3.0Databricks Runtime 7.3我目前正在使用數據湖文件系統中的路徑。這是p = dbutils.fs.ls('dbfs:/databricks-datasets/nyctaxi')print(p) [FileInfo(path='dbfs:/databricks-datasets/nyctaxi/readme_nyctaxi.txt', name='readme_nyctaxi.txt', size=916), FileInfo(path='dbfs:/databricks-datasets/nyctaxi/reference/', name='reference/', size=0), FileInfo(path='dbfs:/databricks-datasets/nyctaxi/taxizone/', name='taxizone/', size=0), FileInfo(path='dbfs:/databricks-datasets/nyctaxi/tripdata/', name='tripdata/', size=0)]現在，為了將其轉換為有效的 Pathlib Posix 對象，我通過函數傳遞它def create_valid_path(paths): return Path('/dbfs').joinpath(*[part for part in Path(paths).parts[1:]])的輸出tripdata是PosixPath('/dbfs/databricks-datasets/nyctaxi/tripdata')現在，如果我想在將 csv 的子集收集到列表中后將其讀入 Sparkdata 框架。from pyspark.sql.functions import * df = spark.read.format('csv').load(paths)這返回AttributeError: 'PosixPath' object has no attribute '_get_object_id'現在，我可以讓它工作的唯一方法是手動添加路徑dbfs:/..并將每個項目返回到字符串，但是有必要使用 Pathlib 來執行一些基本的 I/O 操作。我是否遺漏了一些簡單的東西，或者 Pyspark 根本無法讀取 pathlib 對象？例如trip_paths_str = [str(Path('dbfs:').joinpath(*part.parts[2:])) for part in trip_paths]print(trip_paths_str)['dbfs:/databricks-datasets/nyctaxi/tripdata/fhv/fhv_tripdata_2015-01.csv.gz', 'dbfs:/databricks-datasets/nyctaxi/tripdata/fhv/fhv_tripdata_2015-02.csv.gz'...]
1 回答
慕標琳琳
TA貢獻1830條經驗 獲得超9個贊
那么這樣做怎么樣？
from pyspark.sql.functions import *
import os
def db_list_files(file_path):
file_list = [file.path for file in dbutils.fs.ls(file_path) if os.path.basename(file.path)]
return file_list
files = db_list_files('dbfs:/FileStore/tables/')
df = spark.read.format('text').load(files)
df.show()
添加回答
舉報
0/150
提交
取消