为Spark向前填充。net

Question

我正在查看。NET(C#)中SparkDataFrame的窗口函数。

我有一个DataFramedf，其中列有年、月、日、小时、分钟、ID、类型和值：

| 2021 |  3  |  4  |  8  |  9  |  87  |  Type1  |  380.5  |

| 2021 |  3  |  4  |  8  |  10 | null |   null  |   null  |

| 2021 |  3  |  4  |  8  |  11 | null |   null  |   null  |

| 2021 |  3  |  4  |  8  |  12 | null |   null  |   null  |

| 2021 |  3  |  4  |  8  |  13 |  87  |  Type1  |    0.0  |

| 2021 |  3  |  4  |  8  |  14 |  87  |  Type1  |    0.0  |

我希望根据年、月、日、小时、分钟，用上一行的值填充空行（空行），如下所示：

| 2021 |  3  |  4  |  8  |  9  |  87  |  Type1  |  380.5  |

| 2021 |  3  |  4  |  8  |  10 |  87  |  Type1  |  380.5  |

| 2021 |  3  |  4  |  8  |  11 |  87  |  Type1  |  380.5  |

| 2021 |  3  |  4  |  8  |  12 |  87  |  Type1  |  380.5  |

| 2021 |  3  |  4  |  8  |  13 |  87  |  Type1  |    0.0  |

| 2021 |  3  |  4  |  8  |  14 |  87  |  Type1  |    0.0  |

var filledDataFrame=df.withcolumn（“newvalue”,functions.when(df[“value”].isnull(),functions.lag(df[“value”],1).over（window））。否则(df[“value”])

如何在.NET中为Spark定义窗口，并使用Lag函数向前填充空值？

Answer 1

要在Apache Spark中使用Lag和带.NET的窗口，您非常接近，需要：

var spark = SparkSession.Builder().GetOrCreate();
var df = spark.CreateDataFrame(new List<GenericRow>()
{
    new GenericRow(new object[] {2021, 3, 4, 8, 9, 87, "Type1", 380.5}),
    new GenericRow(new object[] {2021, 3, 4, 8, 10, null, null, null}),
    new GenericRow(new object[] {2021, 3, 4, 8, 11, null, null, null}),
    new GenericRow(new object[] {2021, 3, 4, 8, 12, null, null, null}),
    new GenericRow(new object[] {2021, 3, 4, 8, 13, 87, "Type1", 0.0}),
    new GenericRow(new object[] {2021, 3, 4, 8, 14, 87, "Type1", 0.0})
}, new StructType(new List<StructField>()
{
    new StructField("Year", new IntegerType()),
    new StructField("Month", new IntegerType()),
    new StructField("Day", new IntegerType()),
    new StructField("Hour", new IntegerType()),
    new StructField("Minute", new IntegerType()),
    new StructField("ID", new IntegerType()),
    new StructField("Type", new StringType()),
    new StructField("Value", new DoubleType()),

}));

var window = Window.OrderBy("Year", "Month", "Day", "Hour", "Minute");
var filledDataFrame = df.WithColumn("newValue",
    Functions.When(df["Value"].IsNull(),
            Functions.Lag(df["Value"], 1).Over(window))
        .Otherwise(df["Value"]));

filledDataFrame.Show(1000, 10000);

这将导致：

+----+-----+---+----+------+----+-----+-----+--------+
|Year|Month|Day|Hour|Minute|  ID| Type|Value|newValue|
+----+-----+---+----+------+----+-----+-----+--------+
|2021|    3|  4|   8|     9|  87|Type1|380.5|   380.5|
|2021|    3|  4|   8|    10|null| null| null|   380.5|
|2021|    3|  4|   8|    11|null| null| null|    null|
|2021|    3|  4|   8|    12|null| null| null|    null|
|2021|    3|  4|   8|    13|  87|Type1|  0.0|     0.0|
|2021|    3|  4|   8|    14|  87|Type1|  0.0|     0.0|
+----+-----+---+----+------+----+-----+-----+--------+

但是您可能需要last而不是lag，因为您可以跳过空值：

var spark = SparkSession.Builder().GetOrCreate();
var df = spark.CreateDataFrame(new List<GenericRow>()
{
    new GenericRow(new object[] {2021, 3, 4, 8, 9, 87, "Type1", 380.5}),
    new GenericRow(new object[] {2021, 3, 4, 8, 10, null, null, null}),
    new GenericRow(new object[] {2021, 3, 4, 8, 11, null, null, null}),
    new GenericRow(new object[] {2021, 3, 4, 8, 12, null, null, null}),
    new GenericRow(new object[] {2021, 3, 4, 8, 13, 87, "Type1", 0.0}),
    new GenericRow(new object[] {2021, 3, 4, 8, 14, 87, "Type1", 0.0})
}, new StructType(new List<StructField>()
{
    new StructField("Year", new IntegerType()),
    new StructField("Month", new IntegerType()),
    new StructField("Day", new IntegerType()),
    new StructField("Hour", new IntegerType()),
    new StructField("Minute", new IntegerType()),
    new StructField("ID", new IntegerType()),
    new StructField("Type", new StringType()),
    new StructField("Value", new DoubleType()),

}));

var window = Window.OrderBy("Year", "Month", "Day", "Hour", "Minute");
var filledDataFrame = df.WithColumn("newValue",
    Functions.When(df["Value"].IsNull(),
        Functions.Last(df["Value"], true).Over(window))
        .Otherwise(df["Value"]));

filledDataFrame.Show(1000, 10000);

+----+-----+---+----+------+----+-----+-----+--------+
|Year|Month|Day|Hour|Minute|  ID| Type|Value|newValue|
+----+-----+---+----+------+----+-----+-----+--------+
|2021|    3|  4|   8|     9|  87|Type1|380.5|   380.5|
|2021|    3|  4|   8|    10|null| null| null|   380.5|
|2021|    3|  4|   8|    11|null| null| null|   380.5|
|2021|    3|  4|   8|    12|null| null| null|   380.5|
|2021|    3|  4|   8|    13|  87|Type1|  0.0|     0.0|
|2021|    3|  4|   8|    14|  87|Type1|  0.0|     0.0|
+----+-----+---+----+------+----+-----+-----+--------+

using System;
using System.Collections.Generic;
using Microsoft.Spark.Sql;
using Microsoft.Spark.Sql.Expressions;
using Microsoft.Spark.Sql.Types;