获取 CSV 到 Spark 数据框答案

【问题标题】：Get CSV to Spark dataframe获取 CSV 到 Spark 数据框
【发布时间】：2015-04-29 06:43:44
【问题描述】：

我在 Spark 上使用 python，并希望将 csv 放入数据帧。

Spark SQL 的 documentation 奇怪地没有提供 CSV 作为源的解释。

我找到了Spark-CSV，但是我对文档的两个部分有疑问：

"This package can be added to Spark using the --jars command line option. For example, to include it when starting the spark shell: $ bin/spark-shell --packages com.databricks:spark-csv_2.10:1.0.3" 每次我启动 pyspark 或 spark-submit 时，我真的需要添加这个参数吗？看起来很不雅。有没有办法在python中导入而不是每次都重新下载？
df = sqlContext.load(source="com.databricks.spark.csv", header="true", path = "cars.csv")即使我做了上述，这也行不通。这行代码中的“源”参数代表什么？如何简单地在 linux 上加载本地文件，例如“/Spark_Hadoop/spark-1.3.1-bin-cdh4/cars.csv”？

【问题讨论】：

标签： apache-spark pyspark

【解决方案1】：

使用最新版本的 Spark（我相信是 1.4），这变得容易多了。表达式sqlContext.read 给你一个DataFrameReader 实例，带有.csv() 方法：

df = sqlContext.read.csv("/path/to/your.csv")

请注意，您还可以通过将关键字参数 header=True 添加到 .csv() 调用来指示 csv 文件具有标题。还有一些其他选项可用，并在上面的链接中进行了描述。

【讨论】：

当我调用这个命令时，我收到了错误，AttributeError: 'property' object has no attribute 'csv'。实际上，我错过了所有重要的 sqlContext = SQLContext(sc) - 现在可以正常工作

【解决方案2】：

from pyspark.sql.types import StringType
from pyspark import SQLContext
sqlContext = SQLContext(sc)

Employee_rdd = sc.textFile("\..\Employee.csv")
               .map(lambda line: line.split(","))

Employee_df = Employee_rdd.toDF(['Employee_ID','Employee_name'])

Employee_df.show()

【讨论】：

这个答案有几个赞成票，但我不太清楚发生了什么 - 你制作 SQLContext(sc) 并调用那个 sqlContext，然后你什么都不用做......这只是无关紧要代码？当我在 zeppelin 笔记本中使用简单的 csv 文件尝试相同的代码时，出现错误：Traceback (most recent call last): File "/tmp/zeppelin_pyspark-7664300769638364279.py", line 252, in <module> eval(compiledCode) File "<string>", line 1, in <module> AttributeError: 'int' object has no attribute 'map'
请分享您的代码，以获得帮助。我已经在几个实例中使用了代码，我没有遇到问题
您没有显示 sc 是什么。如果它是 SparkContext()，那么您应该在代码示例中显示分配。

【解决方案3】：

对于 Pyspark，假设 csv 文件的第一行包含标题

spark = SparkSession.builder.appName('chosenName').getOrCreate()
df=spark.read.csv('fileNameWithPath', mode="DROPMALFORMED",inferSchema=True, header = True)

【讨论】：

【解决方案4】：

将 csv 文件读入 RDD，然后从原始 RDD 生成 RowRDD。

在第 1 步创建的 RDD 中创建由 StructType 表示的模式，该模式与 RDD 中的 Rows 结构相匹配。

通过 SQLContext 提供的 createDataFrame 方法将 schema 应用于 Rows 的 RDD。

lines = sc.textFile("examples/src/main/resources/people.txt")
parts = lines.map(lambda l: l.split(","))
# Each line is converted to a tuple.
people = parts.map(lambda p: (p[0], p[1].strip()))

# The schema is encoded in a string.
schemaString = "name age"

fields = [StructField(field_name, StringType(), True) for field_name in schemaString.split()]
schema = StructType(fields)

# Apply the schema to the RDD.
schemaPeople = spark.createDataFrame(people, schema)

来源：SPARK PROGRAMMING GUIDE

【讨论】：

这个答案是旧的，新版本的火花有更简单的方法来实现这一点。参考答案stackoverflow.com/a/41638342/187355和stackoverflow.com/a/46539901/187355

【解决方案5】：

如果你不介意额外的包依赖，你可以使用 Pandas 来解析 CSV 文件。它可以很好地处理内部逗号。

依赖关系：

from pyspark import SparkContext
from pyspark.sql import SQLContext
import pandas as pd

一次将整个文件读入 Spark DataFrame：

sc = SparkContext('local','example')  # if using locally
sql_sc = SQLContext(sc)

pandas_df = pd.read_csv('file.csv')  # assuming the file contains a header
# If no header:
# pandas_df = pd.read_csv('file.csv', names = ['column 1','column 2']) 
s_df = sql_sc.createDataFrame(pandas_df)

或者，更注重数据，您可以将数据分块为 Spark RDD，然后是 DF：

chunk_100k = pd.read_csv('file.csv', chunksize=100000)

for chunky in chunk_100k:
    Spark_temp_rdd = sc.parallelize(chunky.values.tolist())
    try:
        Spark_full_rdd += Spark_temp_rdd
    except NameError:
        Spark_full_rdd = Spark_temp_rdd
    del Spark_temp_rdd

Spark_DF = Spark_full_rdd.toDF(['column 1','column 2'])

【讨论】：

createDataFrame 经常给出这样的错误：IllegalArgumentException: "Error while instantiating 'org.apache.spark.sql.hive.HiveSessionState':" ... 有什么经验吗？

【解决方案6】：

在 Spark 2.0 之后，建议使用 Spark Session：

from pyspark.sql import SparkSession
from pyspark.sql import Row

# Create a SparkSession
spark = SparkSession \
    .builder \
    .appName("basic example") \
    .config("spark.some.config.option", "some-value") \
    .getOrCreate()

def mapper(line):
    fields = line.split(',')
    return Row(ID=int(fields[0]), field1=str(fields[1].encode("utf-8")), field2=int(fields[2]), field3=int(fields[3]))

lines = spark.sparkContext.textFile("file.csv")
df = lines.map(mapper)

# Infer the schema, and register the DataFrame as a table.
schemaDf = spark.createDataFrame(df).cache()
schemaDf.createOrReplaceTempView("tablename")

【讨论】：

【解决方案7】：

我遇到了类似的问题。解决方案是添加一个名为“PYSPARK_SUBMIT_ARGS”的环境变量并将其值设置为“--packages com.databricks:spark-csv_2.10:1.4.0 pyspark-shell”。这适用于 Spark 的 Python 交互式 shell。

确保您的 spark-csv 版本与安装的 Scala 版本相匹配。对于 Scala 2.11，它是 spark-csv_2.11，对于 Scala 2.10 或 2.10.5，它是 spark-csv_2.10。

希望它有效。

【讨论】：

【解决方案8】：

基于 Aravind 的回答，但要短得多，例如：

lines = sc.textFile("/path/to/file").map(lambda x: x.split(","))
df = lines.toDF(["year", "month", "day", "count"])

【讨论】：

【解决方案9】：

使用当前的实现（spark 2.X）你不需要添加 packages 参数，你可以使用内置的 csv 实现

此外，作为公认的答案，您不需要创建 rdd 然后强制执行具有 1 个潜在问题的架构

当您读取 csv 时，它会将所有字段标记为字符串，并且当您使用整数列强制实施架构时，您将获得异常。

完成上述操作的更好方法是

 spark.read.format("csv").schema(schema).option("header", "true").load(input_path).show()

【讨论】：