[
https://issues.apache.org/jira/browse/FLINK-3234?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15132289#comment-15132289
]
ASF GitHub Bot commented on FLINK-3234:
---------------------------------------
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/1585#discussion_r51870368
--- Diff:
flink-java/src/main/java/org/apache/flink/api/java/operators/SortPartitionOperator.java
---
@@ -59,6 +62,14 @@ public SortPartitionOperator(DataSet<T> dataSet, String
sortField, Order sortOrd
this.appendSorting(flatOrderKeys, sortOrder);
}
+ public <K> SortPartitionOperator(DataSet<T> dataSet,
Keys.SelectorFunctionKeys<T, K> sortKey, Order sortOrder, String
sortLocationName) {
+ super(dataSet, dataSet.getType());
+ this.sortLocationName = sortLocationName;
+
+ int[] flatOrderKeys = getFlatFields(sortKey);
--- End diff --
This does not work.
`SelectorFunctionKeys` need to be handled completely different. In short,
you need to inject a MapFunction<IN, Tuple2<KEY, IN>> that extracts the key and
emits a Tuple2 with the extracted key and the original input record.
The data set is then sorted on the first tuple field (the key field) and
the records are unwrapped after the sort by another `MapFunction<Tuple2<KEY,
IN>, IN>`. Have a look at how the PartitionOperator handles KeySelectors.
> SortPartition does not support KeySelectorFunctions
> ---------------------------------------------------
>
> Key: FLINK-3234
> URL: https://issues.apache.org/jira/browse/FLINK-3234
> Project: Flink
> Issue Type: Improvement
> Components: DataSet API
> Affects Versions: 1.0.0, 0.10.1
> Reporter: Fabian Hueske
> Assignee: Chiwan Park
> Fix For: 1.0.0
>
>
> The following is not supported by the DataSet API:
> {code}
> DataSet<MyObject> data = ...
> DataSet<MyObject> data.sortPartition(
> new KeySelector<MyObject, Long>() {
> public Long getKey(MyObject v) {
> ...
> }
> },
> Order.ASCENDING);
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
