在 Spark 中读取多个 CSV 文件并制作一个 DataFrame

Question

我正在使用以下代码读取多个 csv 文件并将它们转换为 pandas df，然后将其连接为单个 pandas df。最后再次转换为 spark DataFrame。我想跳过转换为 pandas df 部分，只想拥有 spark DataFrame。

文件路径

 abfss://xxxxxx/abc/year=2021/month=1/dayofmonth=1/hour=1/*.csv
 abfss://xxxxxx/abc/year=2021/month=1/dayofmonth=1/hour=2/*.csv
......

代码

list = []


for month in range(1,3,1):
  for day in range(1,31,1):
    for hour in range(0,24,1):
      file_location = "//xxxxxx/abc/year=2021/month="+str(month)+"/dayofmonth="+str(day)+"/hour="+str(hour)+"/*.csv"    
     
        try : 
          spark_df = spark.read.format("csv").option("header", "true").load(file_location)
          pandas_df = spark_df.toPandas()
          list.append(pandas_df)
    
    
        except AnalysisException as e:
          print(e)

final_pandas_df = pd.concat(list)
df = spark.createDataFrame(final_pandas_df)

Answer 1

您可以加载所有文件并对分区列应用过滤器：

df = spark.read.format("csv").option("header", "true").load("abfss://xxxxxx/abc/").filter(
    'year = 2021 and month between 1 and 2 and day between 1 and 30 and hour between 0 and 23'
)