Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
doujutun3207
flink
提交
1d00cff8
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,发现更多精彩内容 >>
提交
1d00cff8
编写于
7月 08, 2014
作者:
S
Stephan Ewen
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
[FLINK-1005] Add non-object reusing variants of key-grouped iterator.
Clean minor javadoc errors.
上级
76d4a75e
变更
3
展开全部
隐藏空白更改
内联
并排
Showing
3 changed file
with
557 addition
and
0 deletion
+557
-0
flink-runtime/src/main/java/org/apache/flink/runtime/operators/GroupReduceDriver.java
...org/apache/flink/runtime/operators/GroupReduceDriver.java
+2
-0
stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/util/KeyGroupedIteratorImmutable.java
...sphere/pact/runtime/util/KeyGroupedIteratorImmutable.java
+202
-0
stratosphere-runtime/src/test/java/eu/stratosphere/pact/runtime/util/KeyGroupedIteratorImmutableTest.java
...re/pact/runtime/util/KeyGroupedIteratorImmutableTest.java
+353
-0
未找到文件。
flink-runtime/src/main/java/org/apache/flink/runtime/operators/GroupReduceDriver.java
浏览文件 @
1d00cff8
...
...
@@ -51,6 +51,8 @@ public class GroupReduceDriver<IT, OT> implements PactDriver<GroupReduceFunction
private
TypeComparator
<
IT
>
comparator
;
private
boolean
mutableObjectMode
=
false
;
private
volatile
boolean
running
;
// ------------------------------------------------------------------------
...
...
stratosphere-runtime/src/main/java/eu/stratosphere/pact/runtime/util/KeyGroupedIteratorImmutable.java
0 → 100644
浏览文件 @
1d00cff8
/***********************************************************************************************************************
* 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.pact.runtime.util
;
import
java.io.IOException
;
import
java.util.Iterator
;
import
java.util.NoSuchElementException
;
import
eu.stratosphere.api.common.typeutils.TypeComparator
;
import
eu.stratosphere.api.common.typeutils.TypeSerializer
;
import
eu.stratosphere.util.MutableObjectIterator
;
/**
* The KeyValueIterator returns a key and all values that belong to the key (share the same key).
*
*/
public
final
class
KeyGroupedIteratorImmutable
<
E
>
{
private
final
MutableObjectIterator
<
E
>
iterator
;
private
final
TypeSerializer
<
E
>
serializer
;
private
final
TypeComparator
<
E
>
comparator
;
private
ValuesIterator
valuesIterator
;
private
E
lastKeyRecord
;
private
E
lookahead
;
private
boolean
done
;
/**
* Initializes the KeyGroupedIterator. It requires an iterator which returns its result
* sorted by the key fields.
*
* @param iterator An iterator over records, which are sorted by the key fields, in any order.
* @param serializer The serializer for the data type iterated over.
* @param comparator The comparator for the data type iterated over.
*/
public
KeyGroupedIteratorImmutable
(
MutableObjectIterator
<
E
>
iterator
,
TypeSerializer
<
E
>
serializer
,
TypeComparator
<
E
>
comparator
)
{
if
(
iterator
==
null
||
serializer
==
null
||
comparator
==
null
)
{
throw
new
NullPointerException
();
}
this
.
iterator
=
iterator
;
this
.
serializer
=
serializer
;
this
.
comparator
=
comparator
;
}
/**
* Moves the iterator to the next key. This method may skip any values that have not yet been returned by the
* iterator created by the {@link #getValues()} method. Hence, if called multiple times it "removes" key groups.
*
* @return true, if the input iterator has an other group of records with the same key.
*/
public
boolean
nextKey
()
throws
IOException
{
if
(
lookahead
!=
null
)
{
// common case: whole value-iterator was consumed and a new key group is available.
this
.
comparator
.
setReference
(
this
.
lookahead
);
this
.
valuesIterator
.
next
=
this
.
lookahead
;
this
.
lastKeyRecord
=
this
.
lookahead
;
this
.
lookahead
=
null
;
return
true
;
}
// first element, empty/done, or the values iterator was not entirely consumed
if
(
this
.
done
)
{
return
false
;
}
if
(
this
.
valuesIterator
!=
null
)
{
// values was not entirely consumed. move to the next key
// Required if user code / reduce() method did not read the whole value iterator.
E
next
=
this
.
serializer
.
createInstance
();
while
(
true
)
{
if
((
next
=
this
.
iterator
.
next
(
next
))
!=
null
)
{
if
(!
this
.
comparator
.
equalToReference
(
next
))
{
// the keys do not match, so we have a new group. store the current key
this
.
comparator
.
setReference
(
next
);
this
.
valuesIterator
.
next
=
next
;
this
.
lastKeyRecord
=
next
;
return
true
;
}
}
else
{
// input exhausted
this
.
valuesIterator
.
next
=
null
;
this
.
valuesIterator
=
null
;
this
.
lastKeyRecord
=
null
;
this
.
done
=
true
;
return
false
;
}
}
}
else
{
// first element
// get the next element
E
first
=
this
.
iterator
.
next
(
this
.
serializer
.
createInstance
());
if
(
first
!=
null
)
{
this
.
comparator
.
setReference
(
first
);
this
.
valuesIterator
=
new
ValuesIterator
(
first
);
this
.
lastKeyRecord
=
first
;
return
true
;
}
else
{
// empty input, set everything null
this
.
done
=
true
;
return
false
;
}
}
}
private
E
advanceToNext
()
{
try
{
E
next
=
this
.
iterator
.
next
(
serializer
.
createInstance
());
if
(
next
!=
null
)
{
if
(
comparator
.
equalToReference
(
next
))
{
// same key
return
next
;
}
else
{
// moved to the next key, no more values here
this
.
lookahead
=
next
;
return
null
;
}
}
else
{
// backing iterator is consumed
this
.
done
=
true
;
return
null
;
}
}
catch
(
IOException
e
)
{
throw
new
RuntimeException
(
"An error occurred while reading the next record."
,
e
);
}
}
public
E
getCurrent
()
{
return
lastKeyRecord
;
}
public
TypeComparator
<
E
>
getComparatorWithCurrentReference
()
{
return
this
.
comparator
;
}
/**
* Returns an iterator over all values that belong to the current key. The iterator is initially <code>null</code>
* (before the first call to {@link #nextKey()} and after all keys are consumed. In general, this method returns
* always a non-null value, if a previous call to {@link #nextKey()} return <code>true</code>.
*
* @return Iterator over all values that belong to the current key.
*/
public
ValuesIterator
getValues
()
{
return
this
.
valuesIterator
;
}
// --------------------------------------------------------------------------------------------
public
final
class
ValuesIterator
implements
Iterator
<
E
>
{
private
E
next
;
private
ValuesIterator
(
E
first
)
{
this
.
next
=
first
;
}
@Override
public
boolean
hasNext
()
{
return
next
!=
null
;
}
@Override
public
E
next
()
{
if
(
this
.
next
!=
null
)
{
E
current
=
this
.
next
;
this
.
next
=
KeyGroupedIteratorImmutable
.
this
.
advanceToNext
();
return
current
;
}
else
{
throw
new
NoSuchElementException
();
}
}
@Override
public
void
remove
()
{
throw
new
UnsupportedOperationException
();
}
}
}
stratosphere-runtime/src/test/java/eu/stratosphere/pact/runtime/util/KeyGroupedIteratorImmutableTest.java
0 → 100644
浏览文件 @
1d00cff8
此差异已折叠。
点击以展开。
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录