如何通过PySpark从Blob存储的文本/Excel文件提取指定列创建DataFrame

一、处理文本文件（以分隔符分隔的TXT为例） #

假设你的file1.txt是用逗号、制表符等分隔的结构化文本（带表头），可按以下步骤操作：

1. 初始化SparkSession
2. 读取文本文件并识别表头
3. 选择目标列b、c、d

代码示例：

from pyspark.sql import SparkSession

# 初始化Spark会话
spark = SparkSession.builder.appName("ExtractTxtColumns").getOrCreate()

# 读取文本文件，根据实际分隔符调整sep参数（逗号用","，制表符用"\t"）
df = spark.read.csv("/mnt/reservoir/files/file1.txt", header=True, sep=",")

# 提取指定列
target_df = df.select("b", "c", "d")

# 验证结果
target_df.show()

如果文件没有表头，则需通过列索引选择（假设原文件列顺序为a、b、c、d、e，对应索引0到4）：

from pyspark.sql.functions import col

df = spark.read.csv("/mnt/reservoir/files/file1.txt", header=False, sep=",")
target_df = df.select(col("_c1"), col("_c2"), col("_c3"))
# 可选：给列重命名
target_df = target_df.withColumnRenamed("_c1", "b").withColumnRenamed("_c2", "c").withColumnRenamed("_c3", "d")

二、处理Excel文件 #

读取Excel文件需要依赖第三方库spark-excel，需先配置依赖，再执行读取和列选择：

1. 初始化SparkSession并添加依赖
2. 读取Excel文件并识别表头
3. 提取目标列

代码示例：

from pyspark.sql import SparkSession

# 初始化Spark会话，添加spark-excel依赖（版本需匹配你的Spark和Scala版本）
spark = SparkSession.builder.appName("ExtractExcelColumns") \
    .config("spark.jars.packages", "com.crealytics:spark-excel_2.12:0.13.7") \
    .getOrCreate()

# 读取Excel文件，指定表头；若文件是.xlsx后缀，路径需对应调整
df = spark.read.format("com.crealytics.spark.excel") \
    .option("header", "true") \
    .option("inferSchema", "true")  # 可选，自动推断列类型
    .load("/mnt/reservoir/files/file1.xlsx")  # 注意：Excel文件通常后缀为.xls/.xlsx，需确认路径正确性

# 提取指定列
target_df = df.select("b", "c", "d")

# 验证结果
target_df.show()

注意事项：

• 依赖版本需适配你的Spark版本（比如Spark 3.x对应Scala 2.12，选择匹配的spark-excel版本）
• 如果Excel有多个工作表，可通过option("sheetName", "Sheet1")指定目标工作表

内容的提问来源于stack exchange，提问作者Swati B