使用 .csv 格式的 HDFS 文件创建 Pandas DataFrame

Posted 2023-04-15

【中文标题】使用 .csv 格式的 HDFS 文件创建 Pandas DataFrame

【英文标题】：Creating a Pandas DataFrame with HDFS file in .csv format

【发布时间】：2016-09-21 18:21:00

【问题描述】：

我试图通过从 hadoop 集群中获取 .csv 数据并将其放入 Pandas DataFrame 来创建 Spark 工作流。我能够从 HDFS 中提取数据并将其放入 RDD，但无法将其处理到 Pandas Dataframe 中。 以下是我的代码：

import pandas as pd
import numpy as nm

A=sc.textFile("hdfs://localhost:9000/sales_ord_univ.csv") # this creates the RDD
B=pd.DataFrame(A) # this gives me the following error:pandas.core.common.PandasError: DataFrame constructor not properly called!

我很确定这个错误是由于 RDD 是一个大的单个 列表， 因此我尝试通过';'分割数据（即每个新行都是不同的字符串） 但这似乎也无济于事。

我的总体目标是使用 Pandas 将 CSV 转换为 JSON 并输出到 MongoDB。我已经使用 DictReader、PysparkSQL 完成了这个项目，但想检查是否可以使用 Pandas。

任何帮助将不胜感激 谢谢！

【参考方案1】：

我建议将 csv 加载到 Spark DataFrame 中并将其转换为 Pandas DataFrame。

csvDf = sqlContext.read.format("csv").option("header", "true").option("inferschema", "true").option("mode", "DROPMALFORMED").load("hdfs://localhost:9000/sales_ord_univ.csv") 
B = csvDf.toPandas()

如果您仍在使用小于 2.0 的 Spark 版本，则必须使用 read.format("com.databricks.spark.csv") 并包含 com.databricks.spark.csv 包（例如，使用 pyspark shell 时带有 --packages 参数）。

【参考方案2】：

你需要hdfs (2.0.16)

from hdfs import Config
zzodClient = Config().get_client('zzod') #refer to the docs to set up config
with zzodClient.read(q2Path) as r2Reader:
    r2 = pandas.read_csv(r2Reader)