如何使用Spark从本地驱动节点读取csv文件？

Question

我必须将Amazon S3中的文件解压缩到我的驱动程序节点（Spark集群）中，并且我需要将所有这些csv文件作为Spark Dataframe加载，但是当我试图从驱动程序节点加载数据时，我发现了下一个问题：

Pyspark：

df = self.spark.read.format("csv").option("header", True).load("file:/databricks/driver/*.csv")

“路径不存在：file://folder/*.csv”

我试图使用dbutils.fs.mv（）将所有这些文件移到dbfs，但我运行的是一个Python文件，不能使用dbutils（）。我想我需要广播该文件，但我不知道如何广播，因为我尝试了self.sc.textfile（“file://databricks/driver/*.csv”）。collect()和self.sc.addfile（“file://databricks/driver/*.csv”）和进程无法找到这些文件。

在运行以下代码时更新：

import os
BaseLogs("INFO", os.getcwd())
folders = []
for r, d, f in os.walk(os.getcwd()):
    for folder in d:
      folders.append(os.path.join(r, folder))

for f in folders:
    BaseLogs("INFO", f)
BaseLogs("INFO", os.listdir("/databricks/driver/zipFiles/s3Sensor/2017/Tracking_Bounces_20190906.csv.zip"))
BaseLogs("INFO", os.listdir("/databricks/driver/zipFiles/s3Sensor/2017/Tracking_Opens_20190907.zip"))

try:
    df = self.spark.read.format("csv").option("header", True).option("inferSchema", "true").load("file:///databricks/driver/zipFiles/s3Sensor/2017/Tracking_Bounces_20190906.csv.zip/Bounces.csv")                       
except Exception as e:
    BaseLogs("INFO", e)
    BaseLogs("INFO", "Reading {0} as Spark Dataframe".format("file://" +  file  + ".csv"))
    df = self.spark.read.format("csv").option("header", True).option("inferSchema", "true").load("file://" + file + ".csv")  

2019-10-24T15:16:25.321+0000:[GC（分配失败）[Psyounggen:470370K->14308K(630272K)]479896K->30452K(886784K),0.0209171秒][时间：user=0.04 sys=0.01,real=0.02秒]2019-10-24T15:16:25.977+0000:[GC（元数据GC阈值）[Psyounggen:211288K->20462K(636416K)]227432K->64316K(892928K),0.0285984秒][时间：user=0.04 sys=0.02,real=0.02秒]2019-10-24T15:16:26.006+0000:[完整GC（元数据GC阈值）[Psyounggen:20462K->0K(636416K)）][paroldgen:43854K->55206K(377344K)]64316K->55206K(1013760K)，[metaspace:58323K->58323K(1099776K)]，0.1093583秒][times:user=0.31 sys=0.02，real=0.12秒]2019-10-24t15:16:28.333+0000:[GC（分配失败）[psyounggen:612077K->23597K(990720K)]667283K->78811K(1368064K)，0.0209207秒[时间:user=0.02sys=0.01，real=0.02秒]信息：调用O195.load时出错。：org.apache.spark.sparkException：由于阶段失败而中止作业：阶段0.0中的任务0失败了4次，最近的失败：阶段0.0中丢失了任务0.3（TID 3,172.31.252.216,executor 0）：java.io.FileNotFoundException：文件文件：/databricks/driver/zipfiles/s3sensor/2017/tracking_bounces_20190906.csv.zip/bounces.csv不存在，可能基础文件已更新。您可以通过在SQL中运行'Refresh TABLE TableName'命令或通过重新创建所涉及的DataSet/DataFrame，显式地使Spark中的缓存无效。在org.apache.spark.sql.execution.datasources.filescanrdd$$anon$1$$anon$2.getNext(filescanrdd.scala:248)在org.html" target="_blank">apache.spark.util.nextiterator.hasNext(nextiterator.scala:73)

Answer 1

您可以尝试将您的数据读入panda Dataframe：

import pandas as pd
pdf = pd.read_csv("file:/databricks/driver/xyz.csv")

并将其转换为spark Dataframe：

df = spark.createDataFrame(pdf)