csv - 如何在 pyspark 中使用模式读取 csv

Question

我知道如何使用 pyspark 读取 csv，但我在以正确格式加载它时遇到了很多问题。我的 csv 有 3 列，其中第一列和第二列是字符串，但第三列是字典列表。我无法加载最后一列。

我试过了

schema = StructType([
    StructField("_id", StringType()),
    StructField("text", StringType()),
    StructField("links", ArrayType(elementType=MapType(StringType(), StringType())))
])

但它引发了一个错误。对于 Inferschema，它都不起作用。

Answer 1

你需要有inferSchema="true". 如果它导致问题，请将所有内容都读取为字符串，然后您可以使用ast.literal_eval()from astpackage 将其转换str为dict.

您使用此功能：

def read_csv_spark(spark, file_path):

    """
    :param spark: SparkSession or SQLContext
    :param file_path: Path to the file
    :return: Spark Dataframe
    """
    df = (
        spark.read.format("com.databricks.spark.csv")
        .options(header="true", inferSchema="true")
        .load(file_path)
    )
    return df