PySpark：如何用逗号指定列作为十进制

Question

我正在使用 PySpark 并加载一个 csv 文件。我有一列包含欧洲格式的数字，这意味着逗号替换了点，反之亦然。

例如：我有2.416,67 而不是2,416.67。

My data in .csv file looks like this -    
ID;    Revenue
21;    2.645,45
23;   31.147,05
.
.
55;    1.009,11

在 pandas 中，通过在 pd.read_csv() 中指定 decimal=',' 和 thousands='.' 选项以读取欧洲格式，可以轻松读取此类文件。

熊猫代码：

import pandas as pd
df=pd.read_csv("filepath/revenues.csv",sep=';',decimal=',',thousands='.')

我不知道如何在 PySpark 中做到这一点。

PySpark 代码：

from pyspark.sql.types import StructType, StructField, FloatType, StringType
schema = StructType([
            StructField("ID", StringType(), True),
            StructField("Revenue", FloatType(), True)
                    ])
df=spark.read.csv("filepath/revenues.csv",sep=';',encoding='UTF-8', schema=schema, header=True)

谁能建议我们如何使用上面提到的.csv() 函数在 PySpark 中加载这样的文件？

Answer 1

由于数据的格式，您将无法将其读取为浮点数。您需要将其读取为字符串，将其清理干净，然后转换为浮点数：

from pyspark.sql.functions import regexp_replace
from pyspark.sql.types import FloatType

df = spark.read.option("headers", "true").option("inferSchema", "true").csv("my_csv.csv", sep=";")
df = df.withColumn('revenue', regexp_replace('revenue', '\\.', ''))
df = df.withColumn('revenue', regexp_replace('revenue', ',', '.'))
df = df.withColumn('revenue', df['revenue'].cast("float"))

您也可以将这些链接在一起：

df = spark.read.option("headers", "true").option("inferSchema", "true").csv("my_csv.csv", sep=";")
df = (
         df
         .withColumn('revenue', regexp_replace('revenue', '\\.', ''))
         .withColumn('revenue', regexp_replace('revenue', ',', '.'))
         .withColumn('revenue', df['revenue'].cast("float"))
     )

请注意，我没有对此进行测试，因此可能有一两个错字。

Answer 2

如果您的数据集有很多浮点列，但数据集的大小仍然足够小，可以先使用 pandas 对其进行预处理，我发现执行以下操作会更容易。

import pandas as pd

df_pandas = pd.read_csv('yourfile.csv', sep=';', decimal=',')
df_pandas.to_csv('yourfile__dot_as_decimal_separator.csv', sep=';', decimal='.') # optionally also header=True of course.

df_spark = spark.csv.read('yourfile__dot_as_decimal_separator.csv', sep=';', inferSchema=True) # optionally also header=True of course.

我确实发现 jhole89 的答案非常有用，但发现将它应用于具有很多列（数百列）的数据集很痛苦。

我的意思是：

• 手动指定浮点列并对其进行转换非常费力，
• 尝试通过检查哪些列是字符串类型并包含逗号来动态查找它们，避免不考虑带有毫秒分隔符的日期时间列等，在某些列上强制转换为浮点数，因为它们是包含文本的逗号，但不打算被解析为浮点数：这会导致头痛。

因此，如果有多个浮点列并且你的数据集可以用pandas进行预处理，你可以应用上面的代码。

Answer 3

确保您的 SQL 表已预先格式化为读取 NUMERIC 而不是 INTEGER。我在试图弄清楚所有关于编码以及点和逗号等的不同格式时遇到了很大的麻烦。最后，问题更加原始，它被预先格式化为只读整数，因此无论使用逗号还是点，都不会接受小数。然后我只需要更改我的 SQL 表来接受实数 (NUMERIC) 就可以了。