我想在 Data Fusion 中对我的数据实现 rank 或分区 row_num 函数，但我没有找到任何插件。
有什么办法吗？
我想实现以下，
假设我有上面的数据，现在我想根据 AccountNumber 对数据进行分组，并将最近的记录发送到一个接收器并休息到其他接收器。 所以从上面的数据来看，
Sink1 预计有，
水槽2，
我计划通过按 AccountNumber 应用 rank 或 row_number 分区并按 Record_date desc 排序类似功能来进行这种隔离，并将 rank=1 或 row_num=1 的记录发送到一个接收器，然后发送到另一个接收器。
【问题讨论】：
标签： google-cloud-data-fusion cdap
解决问题的好方法是使用Spark plugin。
要将其添加到您的 Datafusion 实例，请转到 HUB -> 插件 -> 搜索 Spark -> 部署插件。然后您可以在 Analytics 选项卡上找到它。
为了给您举例说明如何使用它，我创建了以下管道：
这条管道基本上是：
现在让我们更深入地了解每个组件：
1 - GCS：这是一个简单的 GCS 源。本例使用的文件数据如下所示
2 - Spark_rank：这是一个 Spark 插件，代码如下。该代码基本上使用您的数据创建了一个临时视图，并且它们应用查询来对您的行进行排名。之后，您的数据将返回到管道。您还可以在下面看到此步骤的输入和输出数据。请注意输出是重复的，因为它被传递到两个分支。
def transform(df: DataFrame, context: SparkExecutionPluginContext) : DataFrame = {
df.createTempView("source")
df.sparkSession.sql("SELECT AccountNumber, Address, Record_date, RANK() OVER (PARTITION BY accountNumber ORDER BY record_date DESC) as rank FROM source")
}
3 - Spark2 和 Spark3：与下面的步骤一样，此步骤使用 Spark 插件来转换数据。 Spark2 使用下面的代码只获取 rank = 1 的数据
def transform(df: DataFrame, context: SparkExecutionPluginContext) : DataFrame = {
df.createTempView("source_0")
df.sparkSession.sql("SELECT AccountNumber, Address, Record_date FROM
source_0 WHERE rank = 1")
}
Spark3 使用以下代码获取 rank > 1 的数据：
def transform(df: DataFrame, context: SparkExecutionPluginContext) : DataFrame = {
df.createTempView("source_1")
df.sparkSession.sql("SELECT accountNumber, address, record_date FROM source_1 WHERE rank > 1")
}
4 - GCS2 和 GCS3：最后，在这一步中，您的数据会再次保存到 GCS。
【讨论】：