pyspark 批量读取 hdfs 上一个路径里面的所有 csv 文件，并将所有 csv 转化为 df，依次存入一个列表中

在Pyspark中，你可以通过`SparkSession`的`read`函数结合`csv`模式和`HadoopFsRelation`来批量读取HDFS上特定路径下的所有CSV文件。然后，可以使用循环或者`RDD`操作来逐个读取并转换为DataFrame，最后存储在一个列表中。以下是一个示例步骤：

from pyspark.sql import SparkSession

# 创建SparkSession
spark = SparkSession.builder.getOrCreate()

# 指定HDFS路径
hdfs_path = "hdfs://path/to/csv/files"

# 定义一个空列表来存储DataFrame
dfs_list = []

# 使用glob.glob获取路径下所有的csv文件
import glob
csv_files = glob.glob(f"{hdfs_path}/*.csv")

# 循环遍历每个csv文件
for csv_file in csv_files:
    # 读取csv文件
    df = spark.read.csv(csv_file, inferSchema=True, header=True)
    
    # 将DataFrame添加到列表中
    dfs_list.append(df)

# 列表中的所有DataFrames已经准备好了

相关问题

根据dwd层的数据使用spark对数据进行处理计算，并讲结果存入mysql

在大数据处理场景中，Spark是一个常用的大规模数据处理框架，它支持实时流处理和批处理任务。如果你需要基于DWD（描述性维度建模）层的数据（通常是结构化数据），通过Spark进行处理，一般可以按照以下步骤进行：

1. **数据读取**：首先使用Spark SQL或者DataFrame API从Hadoop HDFS、Hive metastore或其他数据源（如CSV、Parquet等）加载DWD层数据。

from pyspark.sql import SparkSession

# 创建SparkSession
spark = SparkSession.builder.appName("DWD Data Processing").getOrCreate()

# 读取DWD数据
df_dwd = spark.read.format("parquet") \
   .option("path", "hdfs://your-dwd-path") \
   .load()

2. **数据清洗和预处理**：对数据进行去重、填充缺失值、转换格式等操作，可能需要用到ForeachWriter或udf（用户定义函数）。

import pyspark.sql.functions as F

# 数据清洗示例
cleaned_df = df_dwd.drop_duplicates() \
    .withColumn("column_to_fill", F.when(df["column"].isNull(), "default_value").otherwise(df["column"]))

3. **计算分析**：执行复杂的聚合操作，例如分组统计、窗口函数，甚至机器学习模型训练。

aggregated_results = cleaned_df.groupBy("group_column") \
    .agg(F.avg("value_column"), F.count("*"))

4. **将结果写入MySQL**：Spark提供`jdbc`库来连接并保存结果到关系数据库，如MySQL。

from pyspark.sql import JDBCWriter

url = "jdbc:mysql://localhost:3306/your_database"
table_name = "results_table"

jdbccfg = {
    "driver": "com.mysql.cj.jdbc.Driver",
    "url": url,
    "dbtable": table_name,
    "user": "username",
    "password": "password"
}

writer = JDBCWriter(options=jdbccfg)
writer.write dataframe=aggregated_results
writer.close()