在Databricks中用Python处理API返回的字节格式CSV数据

你遇到的问题是spark.read.csv()仅接受文件路径或分布式存储路径，无法直接解析字符串格式的CSV内容。以下是两种可行的解决方法：

方法一：通过RDD构建Spark DataFrame #

将解码后的CSV字符串按行分割并转为RDD，再用Spark的CSV读取器解析：

# 假设decoded是已完成解码的CSV字符串
# 按换行符分割并过滤空行
csv_lines = [line.strip() for line in decoded.split('\n') if line.strip()]
# 转为Spark RDD
csv_rdd = spark.sparkContext.parallelize(csv_lines)
# 读取RDD为DataFrame，配置CSV参数
sparkdf = spark.read.option("header", True) \
                   .option("inferSchema", True) \
                   .option("sep", "|") \
                   .option("quote", '"') \
                   .option("ignoreLeadingWhiteSpace", True) \
                   .csv(csv_rdd)

方法二：借助Pandas中转解析 #

用StringIO将字符串包装为类文件对象，通过Pandas解析后转为Spark DataFrame：

import pandas as pd
from io import StringIO

# 将CSV字符串包装为可读取的IO对象
csv_io = StringIO(decoded)
# Pandas解析CSV，配置分隔符、引号等参数
pd_df = pd.read_csv(csv_io, sep='|', header=0, quotechar='"', skipinitialspace=True)
# 转为Spark DataFrame
sparkdf = spark.createDataFrame(pd_df)

适用场景说明 #

• 若数据量较大，优先选择方法一，基于RDD的处理更适配Spark的分布式计算特性
• 数据量较小时，方法二代码更简洁，无需处理RDD相关操作

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