Spark SQL：对于给定的数据帧分区，是否可以按预定义的顺序逐行处理？
我希望根据列（SECURITY_ID）对数据帧（df）进行分区，然后运行df。foreachpartition（自定义函数）。这很好用。
在每个分区内，必须根据列（秩）对数据进行排序。这很好用。
现在，根据顺序，我想为每个分区按顺序处理一行接一行。比如-
基础数据帧：
+-------------+----------+----------+------
ACCOUNT_NO|SECURITY_ID| QUANTITY| RANK|
+-------------+----------+----------+------
32934789| 290X2| -98763| 3|
3S534789| 290X2| 45300| 2|
3FA34789| 290X2| 12763| 1|
00000019| 290X2|-10177400| 4|
92115301| 35G71| 8003| 2|
91615301| 35G71| -2883| 1|
经过分割和排序
+-------------+----------+----------+------
ACCOUNT_NO|SECURITY_ID| QUANTITY| RANK|
+-------------+----------+----------+------
3FA34789| 290X2| 12763| 1|
3S534789| 290X2| 45300| 2|
32934789| 290X2| -98763| 3|
00000019| 290X2|-10177400| 4|
+-------------+----------+----------+------
ACCOUNT_NO|SECURITY_ID| QUANTITY| RANK|
+-------------+----------+----------+------
91615301| 35G71| -2883| 1|
92115301| 35G71| 8003| 2|
让我们考虑一下这个分区
+-------------+----------+----------+------
ACCOUNT_NO|SECURITY_ID| QUANTITY| RANK|
+-------------+----------+----------+------
3FA34789| 290X2| 12763| 1|
3S534789| 290X2| 45300| 2|
32934789| 290X2| -98763| 3|
00000019| 290X2|-10177400| 4|
我需要根据排名按递增顺序逐个处理行。
这在单节点机器上似乎很好。但我看到，在多节点集群上运行时，处理变得混乱。
我如何确保订单得到保证？
共有1个答案
请尝试对安全分区的Datafame执行coalesce（1）和sort（cols:*）操作，以获得按指定列排序的新Datafame/Dataset，所有操作都是按升序进行的。
df.coalesce(1).sort("RANK").foreach(row => process(row))
-
我有一个JSON，如下所示，还有一个JSONPath，例如， 我想在相应的JSON中找到JSONPath的行号。我正在使用Jackson和JayWayPath库，但无法弄清楚，是否有一种方法可以找到基于JSONPath的行号。如果有人能提出一些想法，我们将不胜感激。谢谢。
-
问题内容： 我有一个对象数组，这些对象的属性称为“ CODE”。 如何通过自定义顺序对数组进行排序，例如： 尝试各种方法均未成功。请帮忙。 问题答案： 您可以将函数与函数一起使用。
-
问题内容： 我有一个数组数组： 需要按特定顺序进行： 3452342 5867867 7867867 1231233 我将如何去做？我之前已经对数组进行了排序，并阅读了许多其他文章，但它们始终基于比较（即valueA <valueB）。 感谢帮助。 问题答案： 您可以用来精确指示如何对数组进行排序。在这种情况下，可以在比较函数中使用该数组。 下面的示例使用a 使生活更轻松。 这项工作的关键是使要比
-
问题内容： 我有一个这样的dataFrame，我想每60分钟进行一次分组，然后从06:30开始分组。 我在用： 我得到这个分组： 但我正在寻找这个结果： 我如何告诉该功能以6小时30分开始以一小时为间隔进行分组？ 如果 .groupby（pd.TimeGrouper（freq =‘60Min’）） 无法完成此 操作 ，最好的方法是怎么做？ 致敬并非常感谢 问题答案： 使用会同中的参数。 指定将使时
-
我正在学习mock，我想知道我是否可以使用类似的代码： 验证身份验证是否成功。 服务。AuthenticateUser（用户用户）： 如您所见，login方法返回一个播放器，但是有没有可能告诉Mockito我只想拿回有效的东西？因此，我可以测试身份验证是否成功，例如： ^这种方法目前不起作用，它在存根上进行了失败的测试。 编辑：我尝试了两种不同的方法： 这些规范符合要求，但我不确定它们是否真的很好
-
鉴于这些实体和存储库可以访问DDBB中的数据： 我想为方法getTotalPurchaseAmounts（长customerId）添加缓存，这样，当为客户添加一些购买时，只有该客户的purchasesd被逐出。 相关的依赖项是： 相关配置： 由于spring缓存（和ehcache）逐出的次数受元素或所有条目的限制，我开发的解决方案是通过友好方式创建缓存（每个客户一个），这样我就可以逐出这些缓存。