使用pyspark将一列拆分为多个列而不使用pandas

Question

我的问题是如何将一列分成多列，我不知道为什么df.toPandas()无法工作。例如，我想将'df_test'更改为'df_test2'。我看到很多示例使用了pandas模块，还有其他方法吗？谢谢提前。

df_test = sqlContext.createDataFrame([
(1, '14-Jul-15'),
(2, '14-Jun-15'),
(3, '11-Oct-15'),
], ('id', 'date'))

df_test2

id     day    month    year
1       14     Jul      15
2       14     Jun      15
1       11     Oct      15

- nathanlim45

Answer 1

Spark >= 2.2

您可以跳过使用unix_timestamp和cast函数，而是直接使用to_date或to_timestamp函数：

from pyspark.sql.functions import to_date, to_timestamp

df_test.withColumn("date", to_date("date", "dd-MMM-yy")).show()
## +---+----------+
## | id|      date|
## +---+----------+
## |  1|2015-07-14|
## |  2|2015-06-14|
## |  3|2015-10-11|
## +---+----------+


df_test.withColumn("date", to_timestamp("date", "dd-MMM-yy")).show()
## +---+-------------------+
## | id|               date|
## +---+-------------------+
## |  1|2015-07-14 00:00:00|
## |  2|2015-06-14 00:00:00|
## |  3|2015-10-11 00:00:00|
## +---+-------------------+

然后应用下面展示的其他日期时间函数。

Spark < 2.2

在单个访问中无法派生多个顶级列。您可以使用结构体或集合类型，并像这样使用UDF：

from pyspark.sql.types import StringType, StructType, StructField
from pyspark.sql import Row
from pyspark.sql.functions import udf, col

schema = StructType([
  StructField("day", StringType(), True),
  StructField("month", StringType(), True),
  StructField("year", StringType(), True)
])

def split_date_(s):
    try:
        d, m, y = s.split("-")
        return d, m, y
    except:
        return None

split_date = udf(split_date_, schema)

transformed = df_test.withColumn("date", split_date(col("date")))
transformed.printSchema()

## root
##  |-- id: long (nullable = true)
##  |-- date: struct (nullable = true)
##  |    |-- day: string (nullable = true)
##  |    |-- month: string (nullable = true)
##  |    |-- year: string (nullable = true)

但在 PySpark 中，不仅语言冗长，而且代价高昂。

对于基于日期的转换，您可以简单地使用内置函数：

from pyspark.sql.functions import unix_timestamp, dayofmonth, year, date_format

transformed = (df_test
    .withColumn("ts",
        unix_timestamp(col("date"), "dd-MMM-yy").cast("timestamp"))
    .withColumn("day", dayofmonth(col("ts")).cast("string"))
    .withColumn("month", date_format(col("ts"), "MMM"))
    .withColumn("year", year(col("ts")).cast("string"))
    .drop("ts"))

同样地，您可以使用regexp_extract来拆分日期字符串。

另请参见从Spark DataFrame中的单个列派生多个列

注意:

如果您使用未修补SPARK-11724版本，则在unix_timestamp(...)之后，cast("timestamp")之前需要进行更正。

- zero323

---

不知道这个回答是否适用于Spark 2.x的更新。谢谢！ - Kai

在Spark 2.2中，您可以跳过unix_timestamp，但这是唯一的重大变化。其余部分基本相同，SQL函数是推荐的方法。 - zero323

Answer 2

这里的解决方法是使用pyspark.sql.functions.split()函数。

df = sqlContext.createDataFrame([
(1, '14-Jul-15'),
(2, '14-Jun-15'),
(3, '11-Oct-15'),
], ('id', 'date'))

split_col = pyspark.sql.functions.split(df['date'], '-')
df = df.withColumn('day', split_col.getItem(0))
df = df.withColumn('month', split_col.getItem(1))
df = df.withColumn('year', split_col.getItem(2))
df = df.drop("date")

- Drashti Dobariya