Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
doujutun3207
flink
提交
698bcc6f
F
flink
项目概览
doujutun3207
/
flink
与 Fork 源项目一致
从无法访问的项目Fork
通知
24
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
F
flink
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
698bcc6f
编写于
5月 12, 2014
作者:
M
Markus Holzemer
提交者:
Fabian Hueske
5月 15, 2014
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
Created new SortedGrouping to move the group order from the Grouping
上级
1b6a3acf
变更
3
隐藏空白更改
内联
并排
Showing
3 changed file
with
126 addition
and
39 deletion
+126
-39
stratosphere-java/src/main/java/eu/stratosphere/api/java/operators/Grouping.java
...ain/java/eu/stratosphere/api/java/operators/Grouping.java
+12
-36
stratosphere-java/src/main/java/eu/stratosphere/api/java/operators/ReduceGroupOperator.java
.../stratosphere/api/java/operators/ReduceGroupOperator.java
+4
-3
stratosphere-java/src/main/java/eu/stratosphere/api/java/operators/SortedGrouping.java
...va/eu/stratosphere/api/java/operators/SortedGrouping.java
+110
-0
未找到文件。
stratosphere-java/src/main/java/eu/stratosphere/api/java/operators/Grouping.java
浏览文件 @
698bcc6f
...
...
@@ -14,8 +14,6 @@
**********************************************************************************************************************/
package
eu.stratosphere.api.java.operators
;
import
java.util.Arrays
;
import
eu.stratosphere.api.common.InvalidProgramException
;
import
eu.stratosphere.api.common.operators.Order
;
import
eu.stratosphere.api.java.DataSet
;
...
...
@@ -38,12 +36,9 @@ import eu.stratosphere.api.java.functions.ReduceFunction;
*/
public
class
Grouping
<
T
>
{
private
final
DataSet
<
T
>
dataSet
;
private
final
Keys
<
T
>
keys
;
protected
final
DataSet
<
T
>
dataSet
;
private
int
[]
groupSortKeyPositions
=
null
;
private
Order
[]
groupSortOrders
=
null
;
protected
final
Keys
<
T
>
keys
;
public
Grouping
(
DataSet
<
T
>
set
,
Keys
<
T
>
keys
)
{
if
(
set
==
null
||
keys
==
null
)
{
...
...
@@ -67,14 +62,6 @@ public class Grouping<T> {
return
this
.
keys
;
}
protected
int
[]
getGroupSortKeyPositions
()
{
return
this
.
groupSortKeyPositions
;
}
protected
Order
[]
getGroupSortOrders
()
{
return
this
.
groupSortOrders
;
}
// --------------------------------------------------------------------------------------------
// Operations / Transformations
// --------------------------------------------------------------------------------------------
...
...
@@ -96,9 +83,6 @@ public class Grouping<T> {
* @see DataSet
*/
public
AggregateOperator
<
T
>
aggregate
(
Aggregations
agg
,
int
field
)
{
if
(
this
.
groupSortKeyPositions
!=
null
)
{
throw
new
UnsupportedOperationException
(
"Sorted groups not supported for Aggregation operation at the moment."
);
}
return
new
AggregateOperator
<
T
>(
this
,
agg
,
field
);
}
...
...
@@ -116,9 +100,6 @@ public class Grouping<T> {
* @see DataSet
*/
public
ReduceOperator
<
T
>
reduce
(
ReduceFunction
<
T
>
reducer
)
{
if
(
this
.
groupSortKeyPositions
!=
null
)
{
throw
new
UnsupportedOperationException
(
"Sorted groups not supported for Aggregation operation at the moment."
);
}
return
new
ReduceOperator
<
T
>(
this
,
reducer
);
}
...
...
@@ -150,14 +131,16 @@ public class Grouping<T> {
*
* @param field The Tuple field on which the group is sorted.
* @param order The Order in which the specified Tuple field is sorted.
* @return A Grouping with specified order of group element.
* @return A
Sorted
Grouping with specified order of group element.
*
* @see Tuple
* @see Order
*/
public
Grouping
<
T
>
sortGroup
(
int
field
,
Order
order
)
{
public
Sorted
Grouping
<
T
>
sortGroup
(
int
field
,
Order
order
)
{
int
pos
;
int
[]
groupSortKeyPositions
;
Order
[]
groupSortOrders
;
if
(!
dataSet
.
getType
().
isTupleType
())
{
throw
new
InvalidProgramException
(
"Specifying order keys via field positions is only valid for tuple data types"
);
...
...
@@ -166,20 +149,13 @@ public class Grouping<T> {
throw
new
IllegalArgumentException
(
"Order key out of tuple bounds."
);
}
if
(
this
.
groupSortKeyPositions
==
null
)
{
this
.
groupSortKeyPositions
=
new
int
[
1
];
this
.
groupSortOrders
=
new
Order
[
1
];
pos
=
0
;
}
else
{
int
newLength
=
this
.
groupSortKeyPositions
.
length
+
1
;
this
.
groupSortKeyPositions
=
Arrays
.
copyOf
(
this
.
groupSortKeyPositions
,
newLength
);
this
.
groupSortOrders
=
Arrays
.
copyOf
(
this
.
groupSortOrders
,
newLength
);
pos
=
newLength
-
1
;
}
groupSortKeyPositions
=
new
int
[
1
];
groupSortOrders
=
new
Order
[
1
];
pos
=
0
;
this
.
groupSortKeyPositions
[
pos
]
=
field
;
this
.
groupSortOrders
[
pos
]
=
order
;
return
this
;
groupSortKeyPositions
[
pos
]
=
field
;
groupSortOrders
[
pos
]
=
order
;
return
new
SortedGrouping
<
T
>(
this
.
dataSet
,
this
.
keys
,
groupSortKeyPositions
,
groupSortOrders
)
;
}
// public <K extends Comparable<K>> Grouping<T> sortGroup(KeySelector<T, K> keyExtractor, Order order) {
...
...
stratosphere-java/src/main/java/eu/stratosphere/api/java/operators/ReduceGroupOperator.java
浏览文件 @
698bcc6f
...
...
@@ -148,10 +148,11 @@ public class ReduceGroupOperator<IN, OUT> extends SingleInputUdfOperator<IN, OUT
po
.
setDegreeOfParallelism
(
this
.
getParallelism
());
// set group order
if
(
grouper
.
getGroupSortKeyPositions
()
!=
null
)
{
if
(
grouper
instanceof
SortedGrouping
)
{
SortedGrouping
<
IN
>
sortedGrouper
=
(
SortedGrouping
<
IN
>)
grouper
;
int
[]
sortKeyPositions
=
g
rouper
.
getGroupSortKeyPositions
();
Order
[]
sortOrders
=
g
rouper
.
getGroupSortOrders
();
int
[]
sortKeyPositions
=
sortedG
rouper
.
getGroupSortKeyPositions
();
Order
[]
sortOrders
=
sortedG
rouper
.
getGroupSortOrders
();
Ordering
o
=
new
Ordering
();
for
(
int
i
=
0
;
i
<
sortKeyPositions
.
length
;
i
++)
{
...
...
stratosphere-java/src/main/java/eu/stratosphere/api/java/operators/SortedGrouping.java
0 → 100644
浏览文件 @
698bcc6f
/***********************************************************************************************************************
*
* Copyright (C) 2010-2013 by the Stratosphere project (http://stratosphere.eu)
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
* specific language governing permissions and limitations under the License.
*
**********************************************************************************************************************/
package
eu.stratosphere.api.java.operators
;
import
java.util.Arrays
;
import
eu.stratosphere.api.common.InvalidProgramException
;
import
eu.stratosphere.api.common.operators.Order
;
import
eu.stratosphere.api.java.DataSet
;
import
eu.stratosphere.api.java.functions.GroupReduceFunction
;
/**
* SortedGrouping is an intermediate step for a transformation on a grouped and sorted DataSet.<br/>
* The following transformation can be applied on sorted groups:
* <ul>
* <li>{@link Grouping#reduce(ReduceFunction)},</li>
* </ul>
*
* @param <T> The type of the elements of the sorted and grouped DataSet.
*/
public
class
SortedGrouping
<
T
>
extends
Grouping
<
T
>
{
private
int
[]
groupSortKeyPositions
;
private
Order
[]
groupSortOrders
;
public
SortedGrouping
(
DataSet
<
T
>
set
,
Keys
<
T
>
keys
,
int
[]
groupSortKeyPositions
,
Order
[]
groupSortOrders
)
{
super
(
set
,
keys
);
if
(
groupSortKeyPositions
==
null
||
groupSortKeyPositions
.
length
==
0
||
groupSortOrders
==
null
||
groupSortOrders
.
length
==
0
)
{
throw
new
InvalidProgramException
(
"Key positions and sort orders must be specified in order to create a SortedGrouping."
);
}
this
.
groupSortKeyPositions
=
groupSortKeyPositions
;
this
.
groupSortOrders
=
groupSortOrders
;
}
protected
int
[]
getGroupSortKeyPositions
()
{
return
this
.
groupSortKeyPositions
;
}
protected
Order
[]
getGroupSortOrders
()
{
return
this
.
groupSortOrders
;
}
/**
* Applies a GroupReduce transformation on a grouped and sorted {@link DataSet}.<br/>
* The transformation calls a {@link GroupReduceFunction} for each group of the DataSet.
* A GroupReduceFunction can iterate over all elements of a group and emit any
* number of output elements including none.
*
* @param reducer The GroupReduceFunction that is applied on each group of the DataSet.
* @return A GroupReduceOperator that represents the reduced DataSet.
*
* @see GroupReduceFunction
* @see GroupReduceOperator
* @see DataSet
*/
public
<
R
>
ReduceGroupOperator
<
T
,
R
>
reduceGroup
(
GroupReduceFunction
<
T
,
R
>
reducer
)
{
return
new
ReduceGroupOperator
<
T
,
R
>(
this
,
reducer
);
}
// --------------------------------------------------------------------------------------------
// Group Operations
// --------------------------------------------------------------------------------------------
/**
* Sorts {@link Tuple} elements within a group on the specified field in the specified {@link Order}.</br>
* <b>Note: Only groups of Tuple elements can be sorted.</b><br/>
* Groups can be sorted by multiple fields by chaining {@link #sortGroup(int, Order)} calls.
*
* @param field The Tuple field on which the group is sorted.
* @param order The Order in which the specified Tuple field is sorted.
* @return A SortedGrouping with specified order of group element.
*
* @see Tuple
* @see Order
*/
public
SortedGrouping
<
T
>
sortGroup
(
int
field
,
Order
order
)
{
int
pos
;
if
(!
dataSet
.
getType
().
isTupleType
())
{
throw
new
InvalidProgramException
(
"Specifying order keys via field positions is only valid for tuple data types"
);
}
if
(
field
>=
dataSet
.
getType
().
getArity
())
{
throw
new
IllegalArgumentException
(
"Order key out of tuple bounds."
);
}
int
newLength
=
this
.
groupSortKeyPositions
.
length
+
1
;
this
.
groupSortKeyPositions
=
Arrays
.
copyOf
(
this
.
groupSortKeyPositions
,
newLength
);
this
.
groupSortOrders
=
Arrays
.
copyOf
(
this
.
groupSortOrders
,
newLength
);
pos
=
newLength
-
1
;
this
.
groupSortKeyPositions
[
pos
]
=
field
;
this
.
groupSortOrders
[
pos
]
=
order
;
return
this
;
}
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录