Fix spark_kafka_processor sort

shaurya-nwse · adchia · commit 8ab1229035e3 · 2023-02-17T09:00:35.000-08:00
Signed-off-by: shaurya.rawat &lt;shaurya.rawat@new-work.se&gt;
diff --git a/sdk/python/feast/infra/contrib/spark_kafka_processor.py b/sdk/python/feast/infra/contrib/spark_kafka_processor.py
@@ -131,7 +131,7 @@ def batch_write(row: DataFrame, batch_id: int):
             # Also add a 'created' column.
             rows = (
                 rows.sort_values(
-                    by=self.join_keys + [self.sfv.timestamp_field], ascending=True
+                    by=[*self.join_keys, self.sfv.timestamp_field], ascending=False
                 )
                 .groupby(self.join_keys)
                 .nth(0)

Original file line number	Diff line number	Diff line change
`@@ -131,7 +131,7 @@ def batch_write(row: DataFrame, batch_id: int):`
`131`	`131`	`# Also add a 'created' column.`
`132`	`132`	`rows = (`
`133`	`133`	`rows.sort_values(`
`134`		`- by=self.join_keys + [self.sfv.timestamp_field], ascending=True`
	`134`	`+ by=[*self.join_keys, self.sfv.timestamp_field], ascending=False`
`135`	`135`	`)`
`136`	`136`	`.groupby(self.join_keys)`
`137`	`137`	`.nth(0)`