PySpark ajoute une colonne à un DataFrame à partir d'une colonne TimeStampType

J'ai un DataFrame que ressembler à quelque chose comme ça. Je veux faire fonctionner le jour de la date_time champ.

root
 |-- host: string (nullable = true)
 |-- user_id: string (nullable = true)
 |-- date_time: timestamp (nullable = true)

J'ai essayé d'ajouter une colonne pour extraire la journée. Pour l'instant, mes tentatives ont échoué.

df = df.withColumn("day", df.date_time.getField("day"))

org.apache.spark.sql.AnalysisException: GetField is not valid on fields of type TimestampType;

Cela a également échoué

df = df.withColumn("day", df.select("date_time").map(lambda row: row.date_time.day))

AttributeError: 'PipelinedRDD' object has no attribute 'alias'

Une idée de comment cela peut être fait?

source d'informationauteur Wai Yip Tung

  1. 30

    Vous pouvez utiliser de simples map:

    df.rdd.map(lambda row:
    Row(row.__fields__ + ["day"])(row + (row.date_time.day, ))
)

    Une autre option consiste à enregistrer une fonction et d'exécuter des requêtes SQL:

    sqlContext.registerFunction("day", lambda x: x.day)
sqlContext.registerDataFrameAsTable(df, "df")
sqlContext.sql("SELECT *, day(date_time) as day FROM df")

    Enfin, vous pouvez définir l'udf, comme ceci:

    from pyspark.sql.functions import udf
from pyspark.sql.types import IntegerType

day = udf(lambda date_time: date_time.day, IntegerType())
df.withColumn("day", day(df.date_time))

    MODIFIER:

    En fait, si vous utilisez SQL brut day fonction est déjà définie (au moins dans Spark 1.4) de sorte que vous pouvez omettre l'udf d'enregistrement. Il fournit également un certain nombre de différents date de traitement, y compris les fonctions:

    Il est également possible d'utiliser un simple date des expressions comme:

    current_timestamp() - expr("INTERVAL 1 HOUR")

    Cela signifie que vous pouvez construire relativement requêtes complexes sans la transmission des données à Python. Par exemple:

    df =  sc.parallelize([
    (1, "2016-01-06 00:04:21"),
    (2, "2016-05-01 12:20:00"),
    (3, "2016-08-06 00:04:21")
]).toDF(["id", "ts_"])

now = lit("2016-06-01 00:00:00").cast("timestamp") 
five_months_ago = now - expr("INTERVAL 5 MONTHS")

(df
    # Cast string to timestamp
    # For Spark 1.5 use cast("double").cast("timestamp")
    .withColumn("ts", unix_timestamp("ts_").cast("timestamp"))
    # Find all events in the last five months
    .where(col("ts").between(five_months_ago, now))
    # Find first Sunday after the event
    .withColumn("next_sunday", next_day(col("ts"), "Sun"))
    # Compute difference in days
    .withColumn("diff", datediff(col("ts"), col("next_sunday"))))