一.在二次排序當中的應用
1.1
說到排序當然第一想到的就是sort by和order by這兩者的區別，也分情況。
在算子當中，兩者沒有區別，orderby()調用的也是sort。order by就是sort的別名。
/**
* Returns a new Dataset sorted by the given expressions.
* This is an alias of the `sort` function.
*
* @group typedrel
* @since 2.0.0
*/
@scala.annotation.varargs
def orderBy(sortExprs: Column*): Dataset[T] = sort(sortExprs : _*)
在spark sql語句中，則關系到是否全局排序。
https://spark.apache.org/docs/3.0.0/sql-ref-syntax-qry-select-orderby.html
The ORDER BY clause is used to return the result rows in a sorted manner in the user specified order. Unlike the SORT BY clause, this clause guarantees a total order in the output
ORDER BY子句用於按照用戶指定的順序以排序的方式返回結果行。與SORT BY子句不同，該子句保證輸出中的總順序
1.2
如果只想針對一個字段的排序時，這些貌似都沒有問題。
但是如果想針對多個字段進行二次排序，三次排序呢？
select col1,col2 from table order by name,time
針對以上語句，我想根據名稱以及時間進行排序，在一個spark 集群里運行並沒有得到預期的值。
為什么order by在一個分布式計算里針對兩個以上的字段進行排序達不到預期的效果呢？我覺得是因為，不同的name數據行分布在不同的節點或者分區上，order by只能保證各分區內的效果。查看結果也確實是，不同的name有一段有一個順序的時間值，然后變成了另一個name,過一會兒，也跳會了原來的name區別。
結果就不貼出來了。
網上有過一些二次排序的方案，個人覺得使用distribute最簡單。
DISTRIBUTE BY子句用於根據輸入表達式對數據進行重新分區。
針對上面的order by在分布式環境下不能全局二次排序的情況，DISTRIBUTE BY可完美解決，因為它的作用就是針對某一字段，把相同的數據划分到同一分區。
然后數據在同一個分區了，那么再使用order by 或者sort by進行排序，二次三次排序都沒有問題。
以下sql語句先對同name的划分到同一分區，然后針對name,time進行排序，可以得到預期效果。
select col1,col2 from table distribute name order by name time
二.
distribute by解決mappartition的迭代器里OOM的問題
2.1考慮一個業務場景。
有一個車輛數據，每天需要計算每個車的統計數據。很籠統。大概就是spark在凌晨讀取前一天數據，然后按車輛ID分組進行計算。
第一想到的多半是group by id,然后在一個mapprittion接收一個迭代器進行循環處理。
dsInit.groupByKey(new MapFunction<DataSourceModel, String>() {
@Override
public String call(DataSourceModel dataSourceModel) throws Exception {
return dataSourceModel.getVin();
}
}, Encoders.STRING())
.flatMapGroups(new FlatMapGroupsFunction<String, DataSourceModel, Result>() {
@Override
public Iterator<Result> call(String key, Iterator<DataSourceModel> iterator) throws Exception {
//iterator迭代處理
}
但這樣的風險很明顯，就是OOM的風險。spark等大數據框架最大的特點就在於管道處理。不怕處理的數據量大，也不怕服務器資源少(滿足最低配置)，可以一點一點處理，再匯總，內存放不下就落磁盤。但mappartition的iterator相當於就是把當前分區的數據全部加載到內存當中來處理，如果當前分區數據量過大，那么OOM就是必然的。
2.2 如果當前分區數據量過大，可以使用其它方案。
2.2.1
使用DISTRIBUTE BY將數據按字段進行分區，通過2.1我們已經能夠確認這一點。
在這基礎上，再進行mappartition，同樣是接收到一個iterator進行處理，沒有OOM風險。
經實證，同樣的數據，使用group by加FlatMapGroupsFunction直接OOM，
使用
select col1,col2 from table DISTRIBUTE BY id sort by id,time....
得到一個dataset,再進行MapGroupsFunctiont可有效避免OOM風險。
2.2.2
這里還涉及到一個mappartion真的比map效率高嗎
https://blog.csdn.net/weixin_29531897/article/details/114732360
這篇文章作者經過測試，mappartion並不一定效率高，而且有OOM風險。
在文章里也提到另一個解決方案。
大概是利用scala的iterator進行map.
def mapFunc[T, U](iterator: Iterator[T], f2: T => (U)) = { iterator.map(x => { f2(x) }) }
或者直接在mapprition里的iterator進行iterator.map(x => { //業務邏輯處理 })
很遺憾，經測試，同樣的上述的數據集，這兩種方法都直接OOM。並沒有達到文章里說的效果。
2.2.3
在一些特定的業務場景下可以使用reduceGroups
相當於rdd的reduce,兩兩處理，最終得到一條結果。
reduceGroups(new ReduceFunction<ChargeIteratorMap>() {
@Override
public ChargeIteratorMap call(ChargeIteratorMap v1, ChargeIteratorMap v2) throws Exception {
return service(v1, v2);
}
})
如上述業務場景，可先分組並進行排序，再使用reduceGroups當前數據與前一條數據進行計算累加計算，並把臨時結果通過java bean或者case class或者第三方流水保存。
注意：reduceGroups在只有一條數據的時候不執行。
參考：
https://spark.apache.org/docs/3.0.0/sql-ref-syntax-qry-select-distribute-by.html