小白学苑

1、使用Python的字典类型数据来构建DataFrame

from pyspark.sql.types import ArrayType, StructField, StructType, StringType, IntegerType, DecimalType
from decimal import Decimal

# List
data = [{"Category": 'Category A', "ID": 1, "Value": Decimal(12.40)},
        {"Category": 'Category B', "ID": 2, "Value": Decimal(30.10)},
        {"Category": 'Category C', "ID": 3, "Value": Decimal(100.01)}
       ]

schema = StructType([
    StructField('Category', StringType(), False),
    StructField('ID', IntegerType(), False),
    StructField('Value', DecimalType(scale=2), True)
])

# 创建DataFrame
df = spark.createDataFrame(data, schema)
print(df.schema)
df.show()

执行以上代码，输出结果如下：

StructType(List(StructField(Category,StringType,false),StructField(ID,IntegerType,false),StructField(Value,DecimalType(10,2),true)))
+----------+---+------+
|  Category| ID| Value|
+----------+---+------+
|Category A|  1| 12.40|
|Category B|  2| 30.10|
|Category C|  3|100.01|
+----------+---+------+

2、使用lit 函数增加两个常量列

from pyspark.sql.functions import lit

df1 = df.withColumn('Str_Col1', lit('1')) \
        .withColumn('Str_Col2', lit('2020-08-09'))
df1.show()
print(df1.schema)

执行以上代码，输出结果如下：

+----------+---+------+--------+----------+
|  Category| ID| Value|Str_Col1|  Str_Col2|
+----------+---+------+--------+----------+
|Category A|  1| 12.40|       1|2020-08-09|
|Category B|  2| 30.10|       1|2020-08-09|
|Category C|  3|100.01|       1|2020-08-09|
+----------+---+------+--------+----------+
StructType(List(StructField(Category,StringType,false),StructField(ID,IntegerType,false),StructField(Value,DecimalType(10,2),true),StructField(Str_Col1,StringType,false),StructField(Str_Col2,StringType,false)))

从输出结果可以看出，当前的数据类型分别是： StringType, IntegerType, DecimalType, StringType 和 StringType。

3、使用cast 函数改变列类型

可使用函数DataFrame.cast来转换数据类型。

from pyspark.sql.types import DateType

df1 = df1.withColumn("Str_Col1_Int", df1['Str_Col1'].cast('int')).drop('Str_Col1') \
         .withColumn('Str_Col2_Date', df1['Str_Col2'].cast(DateType())).drop('Str_Col2')
df1.show()
print(df1.schema)

执行以上代码，输出结果如下：

+----------+---+------+------------+-------------+
|  Category| ID| Value|Str_Col1_Int|Str_Col2_Date|
+----------+---+------+------------+-------------+
|Category A|  1| 12.40|           1|   2020-08-09|
|Category B|  2| 30.10|           1|   2020-08-09|
|Category C|  3|100.01|           1|   2020-08-09|
+----------+---+------+------------+-------------+

StructType(List(StructField(Category,StringType,false),StructField(ID,IntegerType,false),StructField(Value,DecimalType(10,2),true),StructField(Str_Col1_Int,IntegerType,true),StructField(Str_Col2_Date,DateType,true)))

可以看出，新增加的两列已经被转换为IntegerType和DateType。

另外，上面的代码中，cast函数的使用方式也不同：一个使用隐式类型字符串 'int'，而另一个使用显式类型DateType。对于后者，需要确保导入该类。