Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
doujutun3207
flink
提交
54c6a0d9
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,发现更多精彩内容 >>
提交
54c6a0d9
编写于
7月 14, 2014
作者:
G
gaborhermann
提交者:
Stephan Ewen
8月 18, 2014
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
[streaming] Updated type serialization
上级
eaf7febb
变更
2
隐藏空白更改
内联
并排
Showing
2 changed file
with
225 addition
and
112 deletion
+225
-112
flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/streamrecord/StreamRecord.java
...stratosphere/streaming/api/streamrecord/StreamRecord.java
+149
-85
flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/api/streamrecord/StreamRecordTest.java
...tosphere/streaming/api/streamrecord/StreamRecordTest.java
+76
-27
未找到文件。
flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/api/streamrecord/StreamRecord.java
浏览文件 @
54c6a0d9
...
...
@@ -74,10 +74,9 @@ public class StreamRecord implements IOReadableWritable, Serializable {
private
int
numOfFields
;
private
int
numOfTuples
;
private
static
final
Class
<?>[]
CLASSES
=
new
Class
<?>[]
{
Tuple1
.
class
,
Tuple2
.
class
,
Tuple3
.
class
,
Tuple4
.
class
,
Tuple5
.
class
,
Tuple6
.
class
,
Tuple7
.
class
,
Tuple8
.
class
,
Tuple9
.
class
,
Tuple10
.
class
,
Tuple11
.
class
,
Tuple12
.
class
,
Tuple13
.
class
,
Tuple14
.
class
,
Tuple15
.
class
,
Tuple16
.
class
,
Tuple17
.
class
,
Tuple18
.
class
,
private
static
final
Class
<?>[]
CLASSES
=
new
Class
<?>[]
{
Tuple1
.
class
,
Tuple2
.
class
,
Tuple3
.
class
,
Tuple4
.
class
,
Tuple5
.
class
,
Tuple6
.
class
,
Tuple7
.
class
,
Tuple8
.
class
,
Tuple9
.
class
,
Tuple10
.
class
,
Tuple11
.
class
,
Tuple12
.
class
,
Tuple13
.
class
,
Tuple14
.
class
,
Tuple15
.
class
,
Tuple16
.
class
,
Tuple17
.
class
,
Tuple18
.
class
,
Tuple19
.
class
,
Tuple20
.
class
,
Tuple21
.
class
,
Tuple22
.
class
};
// TODO implement equals, clone
...
...
@@ -190,8 +189,7 @@ public class StreamRecord implements IOReadableWritable, Serializable {
* @throws NoSuchTupleException
* , NoSuchFieldException
*/
public
Object
getField
(
int
tupleNumber
,
int
fieldNumber
)
throws
NoSuchTupleException
,
NoSuchFieldException
{
public
Object
getField
(
int
tupleNumber
,
int
fieldNumber
)
throws
NoSuchTupleException
,
NoSuchFieldException
{
Tuple
tuple
;
try
{
tuple
=
tupleBatch
.
get
(
tupleNumber
);
...
...
@@ -205,8 +203,7 @@ public class StreamRecord implements IOReadableWritable, Serializable {
}
}
public
Object
getFieldFast
(
int
tupleNumber
,
int
fieldNumber
)
throws
NoSuchTupleException
,
NoSuchFieldException
{
public
Object
getFieldFast
(
int
tupleNumber
,
int
fieldNumber
)
throws
NoSuchTupleException
,
NoSuchFieldException
{
Tuple
tuple
;
try
{
tuple
=
tupleBatch
.
get
(
tupleNumber
);
...
...
@@ -245,8 +242,7 @@ public class StreamRecord implements IOReadableWritable, Serializable {
* , NoSuchFieldException
*/
// TODO: add exception for cast for all getters
public
Boolean
getBoolean
(
int
tupleNumber
,
int
fieldNumber
)
throws
NoSuchTupleException
,
NoSuchFieldException
{
public
Boolean
getBoolean
(
int
tupleNumber
,
int
fieldNumber
)
throws
NoSuchTupleException
,
NoSuchFieldException
{
return
(
Boolean
)
getField
(
tupleNumber
,
fieldNumber
);
}
...
...
@@ -274,8 +270,7 @@ public class StreamRecord implements IOReadableWritable, Serializable {
* @throws NoSuchTupleException
* , NoSuchFieldException
*/
public
Byte
getByte
(
int
tupleNumber
,
int
fieldNumber
)
throws
NoSuchTupleException
,
NoSuchFieldException
{
public
Byte
getByte
(
int
tupleNumber
,
int
fieldNumber
)
throws
NoSuchTupleException
,
NoSuchFieldException
{
return
(
Byte
)
getField
(
tupleNumber
,
fieldNumber
);
}
...
...
@@ -288,8 +283,7 @@ public class StreamRecord implements IOReadableWritable, Serializable {
* @throws NoSuchTupleException
* , NoSuchFieldException
*/
public
Character
getCharacter
(
int
fieldNumber
)
throws
NoSuchTupleException
,
NoSuchFieldException
{
public
Character
getCharacter
(
int
fieldNumber
)
throws
NoSuchTupleException
,
NoSuchFieldException
{
return
getCharacter
(
0
,
fieldNumber
);
}
...
...
@@ -304,8 +298,7 @@ public class StreamRecord implements IOReadableWritable, Serializable {
* @throws NoSuchTupleException
* , NoSuchFieldException
*/
public
Character
getCharacter
(
int
tupleNumber
,
int
fieldNumber
)
throws
NoSuchTupleException
,
NoSuchFieldException
{
public
Character
getCharacter
(
int
tupleNumber
,
int
fieldNumber
)
throws
NoSuchTupleException
,
NoSuchFieldException
{
return
(
Character
)
getField
(
tupleNumber
,
fieldNumber
);
}
...
...
@@ -333,8 +326,7 @@ public class StreamRecord implements IOReadableWritable, Serializable {
* @throws NoSuchTupleException
* , NoSuchFieldException
*/
public
Double
getDouble
(
int
tupleNumber
,
int
fieldNumber
)
throws
NoSuchTupleException
,
NoSuchFieldException
{
public
Double
getDouble
(
int
tupleNumber
,
int
fieldNumber
)
throws
NoSuchTupleException
,
NoSuchFieldException
{
return
(
Double
)
getField
(
tupleNumber
,
fieldNumber
);
}
...
...
@@ -362,8 +354,7 @@ public class StreamRecord implements IOReadableWritable, Serializable {
* @throws NoSuchTupleException
* , NoSuchFieldException
*/
public
Float
getFloat
(
int
tupleNumber
,
int
fieldNumber
)
throws
NoSuchTupleException
,
NoSuchFieldException
{
public
Float
getFloat
(
int
tupleNumber
,
int
fieldNumber
)
throws
NoSuchTupleException
,
NoSuchFieldException
{
return
(
Float
)
getField
(
tupleNumber
,
fieldNumber
);
}
...
...
@@ -391,8 +382,7 @@ public class StreamRecord implements IOReadableWritable, Serializable {
* @throws NoSuchTupleException
* , NoSuchFieldException
*/
public
Integer
getInteger
(
int
tupleNumber
,
int
fieldNumber
)
throws
NoSuchTupleException
,
NoSuchFieldException
{
public
Integer
getInteger
(
int
tupleNumber
,
int
fieldNumber
)
throws
NoSuchTupleException
,
NoSuchFieldException
{
return
(
Integer
)
getField
(
tupleNumber
,
fieldNumber
);
}
...
...
@@ -420,8 +410,7 @@ public class StreamRecord implements IOReadableWritable, Serializable {
* @throws NoSuchTupleException
* , NoSuchFieldException
*/
public
Long
getLong
(
int
tupleNumber
,
int
fieldNumber
)
throws
NoSuchTupleException
,
NoSuchFieldException
{
public
Long
getLong
(
int
tupleNumber
,
int
fieldNumber
)
throws
NoSuchTupleException
,
NoSuchFieldException
{
return
(
Long
)
getField
(
tupleNumber
,
fieldNumber
);
}
...
...
@@ -449,8 +438,7 @@ public class StreamRecord implements IOReadableWritable, Serializable {
* @throws NoSuchTupleException
* , NoSuchFieldException
*/
public
Short
getShort
(
int
tupleNumber
,
int
fieldNumber
)
throws
NoSuchTupleException
,
NoSuchFieldException
{
public
Short
getShort
(
int
tupleNumber
,
int
fieldNumber
)
throws
NoSuchTupleException
,
NoSuchFieldException
{
return
(
Short
)
getField
(
tupleNumber
,
fieldNumber
);
}
...
...
@@ -476,8 +464,7 @@ public class StreamRecord implements IOReadableWritable, Serializable {
* Position of the field in the tuple
* @return value of the field as String
*/
public
String
getString
(
int
tupleNumber
,
int
fieldNumber
)
throws
NoSuchTupleException
,
NoSuchFieldException
{
public
String
getString
(
int
tupleNumber
,
int
fieldNumber
)
throws
NoSuchTupleException
,
NoSuchFieldException
{
return
(
String
)
getField
(
tupleNumber
,
fieldNumber
);
}
...
...
@@ -598,8 +585,7 @@ public class StreamRecord implements IOReadableWritable, Serializable {
* New value
* @throws NoSuchFieldException
*/
public
void
setCharacter
(
int
tupleNumber
,
int
fieldNumber
,
Character
c
)
throws
NoSuchFieldException
{
public
void
setCharacter
(
int
tupleNumber
,
int
fieldNumber
,
Character
c
)
throws
NoSuchFieldException
{
setField
(
tupleNumber
,
fieldNumber
,
c
);
}
...
...
@@ -822,8 +808,7 @@ public class StreamRecord implements IOReadableWritable, Serializable {
* @throws NoSuchTupleException
* , TupleSizeMismatchException
*/
public
void
getTupleInto
(
int
tupleNumber
,
Tuple
tuple
)
throws
NoSuchTupleException
,
TupleSizeMismatchException
{
public
void
getTupleInto
(
int
tupleNumber
,
Tuple
tuple
)
throws
NoSuchTupleException
,
TupleSizeMismatchException
{
if
(
tuple
.
getArity
()
==
numOfFields
)
{
try
{
...
...
@@ -862,8 +847,7 @@ public class StreamRecord implements IOReadableWritable, Serializable {
* @throws NoSuchTupleException
* , TupleSizeMismatchException
*/
public
void
setTuple
(
int
tupleNumber
,
Tuple
tuple
)
throws
NoSuchTupleException
,
TupleSizeMismatchException
{
public
void
setTuple
(
int
tupleNumber
,
Tuple
tuple
)
throws
NoSuchTupleException
,
TupleSizeMismatchException
{
if
(
tuple
.
getArity
()
==
numOfFields
)
{
try
{
tupleBatch
.
set
(
tupleNumber
,
copyTuple
(
tuple
));
...
...
@@ -978,37 +962,94 @@ public class StreamRecord implements IOReadableWritable, Serializable {
}
/**
* Converts tuple field types to
long
* Converts tuple field types to
a byte array
*
* @param tuple
* Tuple to convert
* @return long number representing all the tuple field types
* @return byte array representing types
*/
private
long
tupleBasicTypesToLong
(
Tuple
tuple
)
{
@SuppressWarnings
(
"rawtypes"
)
int
[]
typeNums
=
new
int
[
numOfFields
];
byte
[]
tupleBasicTypesToByteArray
(
Tuple
tuple
)
{
byte
[]
typeNums
=
new
byte
[
numOfFields
];
for
(
int
i
=
0
;
i
<
typeNums
.
length
;
i
++)
{
String
typeName
=
tuple
.
getField
(
i
).
getClass
().
getName
();
if
(
type
Name
.
equals
(
"java.lang.Boolean"
))
{
Class
<?
extends
Object
>
type
=
tuple
.
getField
(
i
).
getClass
();
if
(
type
.
equals
(
Boolean
.
class
))
{
typeNums
[
i
]
=
0
;
}
else
if
(
type
Name
.
equals
(
"java.lang.Byte"
))
{
}
else
if
(
type
.
equals
(
Byte
.
class
))
{
typeNums
[
i
]
=
1
;
}
else
if
(
type
Name
.
equals
(
"java.lang.Character"
))
{
}
else
if
(
type
.
equals
(
Character
.
class
))
{
typeNums
[
i
]
=
2
;
}
else
if
(
type
Name
.
equals
(
"java.lang.Double"
))
{
}
else
if
(
type
.
equals
(
Double
.
class
))
{
typeNums
[
i
]
=
3
;
}
else
if
(
type
Name
.
equals
(
"java.lang.Float"
))
{
}
else
if
(
type
.
equals
(
Float
.
class
))
{
typeNums
[
i
]
=
4
;
}
else
if
(
type
Name
.
equals
(
"java.lang.Integer"
))
{
}
else
if
(
type
.
equals
(
Integer
.
class
))
{
typeNums
[
i
]
=
5
;
}
else
if
(
type
Name
.
equals
(
"java.lang.Long"
))
{
}
else
if
(
type
.
equals
(
Long
.
class
))
{
typeNums
[
i
]
=
6
;
}
else
if
(
type
Name
.
equals
(
"java.lang.Short"
))
{
}
else
if
(
type
.
equals
(
Short
.
class
))
{
typeNums
[
i
]
=
7
;
}
else
if
(
type
Name
.
equals
(
"java.lang.String"
))
{
}
else
if
(
type
.
equals
(
String
.
class
))
{
typeNums
[
i
]
=
8
;
}
}
return
typeNums
;
}
/**
* Gets tuple field types from a byte array
*
* @param byte array representing types
* @param numberOfFields
* @return Class array of field types
*/
@SuppressWarnings
(
"rawtypes"
)
Class
[]
tupleBasicTypesFromByteArray
(
byte
[]
representation
,
int
numberOfFields
)
{
Class
[]
basicTypes
=
new
Class
[
representation
.
length
];
for
(
int
i
=
0
;
i
<
basicTypes
.
length
;
i
++)
{
switch
(
representation
[
i
])
{
case
0
:
basicTypes
[
i
]
=
java
.
lang
.
Boolean
.
class
;
break
;
case
1
:
basicTypes
[
i
]
=
java
.
lang
.
Byte
.
class
;
break
;
case
2
:
basicTypes
[
i
]
=
java
.
lang
.
Character
.
class
;
break
;
case
3
:
basicTypes
[
i
]
=
java
.
lang
.
Double
.
class
;
break
;
case
4
:
basicTypes
[
i
]
=
java
.
lang
.
Float
.
class
;
break
;
case
5
:
basicTypes
[
i
]
=
java
.
lang
.
Integer
.
class
;
break
;
case
6
:
basicTypes
[
i
]
=
java
.
lang
.
Long
.
class
;
break
;
case
7
:
basicTypes
[
i
]
=
java
.
lang
.
Short
.
class
;
break
;
case
8
:
basicTypes
[
i
]
=
java
.
lang
.
String
.
class
;
break
;
default
:
basicTypes
[
i
]
=
java
.
lang
.
String
.
class
;
break
;
}
}
return
basicTypes
;
}
/**
* Converts tuple field types to long
*
* @param tuple
* Tuple to convert
* @return long number representing all the tuple field types
*/
long
tupleBasicTypesToLong
(
Tuple
tuple
)
{
byte
[]
typeNums
=
tupleBasicTypesToByteArray
(
tuple
);
long
representation
=
0
;
long
multiplyBy
=
1
;
...
...
@@ -1020,7 +1061,6 @@ public class StreamRecord implements IOReadableWritable, Serializable {
return
representation
;
}
// TODO: Test this
/**
* Converts long type representation to Class array
*
...
...
@@ -1028,18 +1068,17 @@ public class StreamRecord implements IOReadableWritable, Serializable {
* Long representation of types
* @param numberOfFields
* The number of fields
* @return
A
rray of field types
* @return
a
rray of field types
*/
private
Class
[]
tupleBasicTypesFromLong
(
long
representation
,
int
numberOfFields
)
{
@SuppressWarnings
(
"rawtypes"
)
Class
[]
tupleBasicTypesFromLong
(
long
representation
,
int
numberOfFields
)
{
int
[]
typeNums
=
new
int
[
numberOfFields
];
long
multiplyBy
=
1
;
for
(
int
i
=
0
;
i
<
typeNums
.
length
;
i
++)
{
typeNums
[
i
]
=
(
int
)
(
representation
%
9
);
representation
=
representation
/
9
;
}
@SuppressWarnings
(
"rawtypes"
)
Class
[]
basicTypes
=
new
Class
[
typeNums
.
length
];
for
(
int
i
=
0
;
i
<
basicTypes
.
length
;
i
++)
{
switch
(
typeNums
[
i
])
{
...
...
@@ -1078,6 +1117,50 @@ public class StreamRecord implements IOReadableWritable, Serializable {
return
basicTypes
;
}
/**
* Converts tuple field types to String
*
* @param tuple
* @return String representing all the tuple field types
*/
@SuppressWarnings
(
"rawtypes"
)
String
tupleBasicTypesToString
(
Tuple
tuple
)
{
Class
[]
basicTypes
=
new
Class
[
tuple
.
getArity
()];
StringBuilder
basicTypeNames
=
new
StringBuilder
();
// TODO: exception for empty record - no getField!
for
(
int
i
=
0
;
i
<
basicTypes
.
length
;
i
++)
{
basicTypes
[
i
]
=
tuple
.
getField
(
i
).
getClass
();
basicTypeNames
.
append
(
basicTypes
[
i
].
getName
()
+
","
);
}
return
basicTypeNames
.
toString
();
}
/**
* Converts tuple field types from String
*
* @param String representation of types
* @param numberOfFields
* @return array of field types
*/
@SuppressWarnings
(
"rawtypes"
)
Class
[]
tupleBasicTypesFromString
(
String
representation
,
int
numberOfFields
)
{
// TODO: use StringTokenizer
String
[]
types
=
representation
.
split
(
","
);
@SuppressWarnings
(
"rawtypes"
)
Class
[]
basicTypes
=
new
Class
[
types
.
length
];
for
(
int
i
=
0
;
i
<
types
.
length
;
i
++)
{
try
{
basicTypes
[
i
]
=
Class
.
forName
(
types
[
i
]);
}
catch
(
ClassNotFoundException
e
)
{
e
.
printStackTrace
();
}
}
return
basicTypes
;
}
/**
* Writes tuple to the specified DataOutput
*
...
...
@@ -1087,31 +1170,23 @@ public class StreamRecord implements IOReadableWritable, Serializable {
* Output chosen
*/
private
void
writeTuple
(
Tuple
tuple
,
DataOutput
out
)
{
@SuppressWarnings
(
"rawtypes"
)
// TODO: exception for empty record - no getField
Class
[]
basicTypes
=
new
Class
[
tuple
.
getArity
()];
StringBuilder
basicTypeNames
=
new
StringBuilder
();
// TODO: exception for empty record - no getField!
for
(
int
i
=
0
;
i
<
basicTypes
.
length
;
i
++)
{
basicTypes
[
i
]
=
tuple
.
getField
(
i
).
getClass
();
basicTypeNames
.
append
(
basicTypes
[
i
].
getName
()
+
","
);
}
TypeInformation
<?
extends
Tuple
>
typeInfo
=
TupleTypeInfo
.
getBasicTupleTypeInfo
(
basicTypes
);
StringValue
typeVal
=
new
StringValue
(
basicTypeNames
.
toString
());
TypeInformation
<?
extends
Tuple
>
typeInfo
=
TupleTypeInfo
.
getBasicTupleTypeInfo
(
basicTypes
);
@SuppressWarnings
(
"unchecked"
)
TupleSerializer
<
Tuple
>
tupleSerializer
=
(
TupleSerializer
<
Tuple
>)
typeInfo
.
createSerializer
();
SerializationDelegate
<
Tuple
>
serializationDelegate
=
new
SerializationDelegate
<
Tuple
>(
tupleSerializer
);
TupleSerializer
<
Tuple
>
tupleSerializer
=
(
TupleSerializer
<
Tuple
>)
typeInfo
.
createSerializer
();
SerializationDelegate
<
Tuple
>
serializationDelegate
=
new
SerializationDelegate
<
Tuple
>(
tupleSerializer
);
serializationDelegate
.
setInstance
(
tuple
);
try
{
typeVal
.
write
(
out
);
out
.
writeInt
(
numOfFields
);
out
.
write
(
tupleBasicTypesToByteArray
(
getTuple
()));
serializationDelegate
.
write
(
out
);
}
catch
(
IOException
e
)
{
// TODO Auto-generated catch block
e
.
printStackTrace
();
}
}
...
...
@@ -1126,25 +1201,14 @@ public class StreamRecord implements IOReadableWritable, Serializable {
*/
private
Tuple
readTuple
(
DataInput
in
)
throws
IOException
{
StringValue
typeVal
=
new
StringValue
();
typeVal
.
read
(
in
);
// TODO: use StringTokenizer
String
[]
types
=
typeVal
.
getValue
().
split
(
","
);
int
numberOfFields
=
in
.
readInt
();
byte
[]
typesInByte
=
new
byte
[
numberOfFields
];
in
.
readFully
(
typesInByte
,
0
,
numberOfFields
);
@SuppressWarnings
(
"rawtypes"
)
Class
[]
basicTypes
=
new
Class
[
types
.
length
];
for
(
int
i
=
0
;
i
<
types
.
length
;
i
++)
{
try
{
basicTypes
[
i
]
=
Class
.
forName
(
types
[
i
]);
}
catch
(
ClassNotFoundException
e
)
{
// TODO Auto-generated catch block
e
.
printStackTrace
();
}
}
Class
[]
basicTypes
=
tupleBasicTypesFromByteArray
(
typesInByte
,
numberOfFields
);
TypeInformation
<?
extends
Tuple
>
typeInfo
=
TupleTypeInfo
.
getBasicTupleTypeInfo
(
basicTypes
);
@SuppressWarnings
(
"unchecked"
)
TupleSerializer
<
Tuple
>
tupleSerializer
=
(
TupleSerializer
<
Tuple
>)
typeInfo
.
createSerializer
();
TupleSerializer
<
Tuple
>
tupleSerializer
=
(
TupleSerializer
<
Tuple
>)
typeInfo
.
createSerializer
();
DeserializationDelegate
<
Tuple
>
dd
=
new
DeserializationDelegate
<
Tuple
>(
tupleSerializer
);
dd
.
setInstance
(
tupleSerializer
.
createInstance
());
...
...
flink-addons/flink-streaming/src/test/java/eu/stratosphere/streaming/api/streamrecord/StreamRecordTest.java
浏览文件 @
54c6a0d9
...
...
@@ -31,18 +31,19 @@ import org.junit.Test;
import
eu.stratosphere.api.java.tuple.Tuple1
;
import
eu.stratosphere.api.java.tuple.Tuple2
;
import
eu.stratosphere.api.java.tuple.Tuple20
;
import
eu.stratosphere.api.java.tuple.Tuple3
;
import
eu.stratosphere.api.java.tuple.Tuple4
;
import
eu.stratosphere.api.java.tuple.Tuple5
;
import
eu.stratosphere.api.java.tuple.Tuple9
;
import
eu.stratosphere.types.StringValue
;
public
class
StreamRecordTest
{
@Test
public
void
singleRecordSetGetTest
()
{
StreamRecord
record
=
new
StreamRecord
(
new
Tuple9
<
String
,
Integer
,
Long
,
Boolean
,
Double
,
Byte
,
Character
,
Float
,
Short
>(
"Stratosphere"
,
1
,
2L
,
true
,
3.5
,
(
byte
)
0xa
,
'a'
,
0.1f
,
(
short
)
42
));
new
Tuple9
<
String
,
Integer
,
Long
,
Boolean
,
Double
,
Byte
,
Character
,
Float
,
Short
>(
"Stratosphere"
,
1
,
2L
,
true
,
3.5
,
(
byte
)
0xa
,
'a'
,
0.1f
,
(
short
)
42
));
assertEquals
(
9
,
record
.
getNumOfFields
());
assertEquals
(
1
,
record
.
getNumOfTuples
());
...
...
@@ -57,8 +58,7 @@ public class StreamRecordTest {
assertEquals
((
Float
)
0.1f
,
record
.
getFloat
(
7
));
assertEquals
((
Short
)
(
short
)
42
,
record
.
getShort
(
8
));
Tuple9
<
String
,
Integer
,
Long
,
Boolean
,
Double
,
Byte
,
Character
,
Float
,
Short
>
tuple
=
new
Tuple9
<
String
,
Integer
,
Long
,
Boolean
,
Double
,
Byte
,
Character
,
Float
,
Short
>();
Tuple9
<
String
,
Integer
,
Long
,
Boolean
,
Double
,
Byte
,
Character
,
Float
,
Short
>
tuple
=
new
Tuple9
<
String
,
Integer
,
Long
,
Boolean
,
Double
,
Byte
,
Character
,
Float
,
Short
>();
record
.
getTupleInto
(
tuple
);
...
...
@@ -111,7 +111,6 @@ public class StreamRecordTest {
assertEquals
((
Character
)
'\0'
,
record
.
getCharacter
(
0
,
6
));
assertEquals
((
Float
)
0
.
f
,
record
.
getFloat
(
0
,
7
));
assertEquals
((
Short
)
(
short
)
0
,
record
.
getShort
(
0
,
8
));
}
...
...
@@ -152,8 +151,7 @@ public class StreamRecordTest {
assertEquals
(
false
,
record
.
getBoolean
(
1
,
3
));
assertEquals
((
Double
)
0
.,
record
.
getDouble
(
1
,
4
));
record
.
setTuple
(
new
Tuple5
<
String
,
Integer
,
Long
,
Boolean
,
Double
>(
""
,
0
,
0L
,
false
,
0
.));
record
.
setTuple
(
new
Tuple5
<
String
,
Integer
,
Long
,
Boolean
,
Double
>(
""
,
0
,
0L
,
false
,
0
.));
assertEquals
(
5
,
record
.
getNumOfFields
());
assertEquals
(
2
,
record
.
getNumOfTuples
());
...
...
@@ -164,8 +162,7 @@ public class StreamRecordTest {
assertEquals
(
false
,
record
.
getBoolean
(
0
,
3
));
assertEquals
((
Double
)
0
.,
record
.
getDouble
(
0
,
4
));
record
.
setTuple
(
1
,
new
Tuple5
<
String
,
Integer
,
Long
,
Boolean
,
Double
>(
"Stratosphere"
,
1
,
2L
,
true
,
3.5
));
record
.
setTuple
(
1
,
new
Tuple5
<
String
,
Integer
,
Long
,
Boolean
,
Double
>(
"Stratosphere"
,
1
,
2L
,
true
,
3.5
));
assertEquals
(
"Stratosphere"
,
record
.
getString
(
1
,
0
));
assertEquals
((
Integer
)
1
,
record
.
getInteger
(
1
,
1
));
...
...
@@ -183,8 +180,7 @@ public class StreamRecordTest {
assertEquals
(
false
,
record
.
getBoolean
(
0
,
3
));
assertEquals
((
Double
)
0
.,
record
.
getDouble
(
0
,
4
));
record
.
addTuple
(
0
,
new
Tuple5
<
String
,
Integer
,
Long
,
Boolean
,
Double
>(
"Stratosphere"
,
1
,
2L
,
true
,
3.5
));
record
.
addTuple
(
0
,
new
Tuple5
<
String
,
Integer
,
Long
,
Boolean
,
Double
>(
"Stratosphere"
,
1
,
2L
,
true
,
3.5
));
assertEquals
(
2
,
record
.
getNumOfTuples
());
...
...
@@ -231,9 +227,8 @@ public class StreamRecordTest {
final
int
ITERATION
=
10000
;
StreamRecord
record
=
new
StreamRecord
(
new
Tuple4
<
Integer
,
Long
,
String
,
String
>(
0
,
42L
,
"Stratosphere"
,
"Streaming"
));
StreamRecord
record
=
new
StreamRecord
(
new
Tuple4
<
Integer
,
Long
,
String
,
String
>(
0
,
42L
,
"Stratosphere"
,
"Streaming"
));
long
t
=
System
.
nanoTime
();
for
(
int
i
=
0
;
i
<
ITERATION
;
i
++)
{
...
...
@@ -252,12 +247,9 @@ public class StreamRecordTest {
StreamRecord
record20
=
new
StreamRecord
(
new
Tuple20
<
Integer
,
Long
,
String
,
String
,
String
,
String
,
String
,
String
,
String
,
String
,
String
,
String
,
String
,
String
,
String
,
String
,
String
,
String
,
String
,
String
>(
0
,
42L
,
"Stratosphere"
,
"Streaming"
,
"Stratosphere"
,
"Stratosphere"
,
"Streaming"
,
"Stratosphere"
,
"Streaming"
,
"Streaming"
,
"Stratosphere"
,
"Streaming"
,
"Stratosphere"
,
"Streaming"
,
"Streaming"
,
"Stratosphere"
,
"Streaming"
,
"Stratosphere"
,
"Streaming"
,
"Streaming"
));
0
,
42L
,
"Stratosphere"
,
"Streaming"
,
"Stratosphere"
,
"Stratosphere"
,
"Streaming"
,
"Stratosphere"
,
"Streaming"
,
"Streaming"
,
"Stratosphere"
,
"Streaming"
,
"Stratosphere"
,
"Streaming"
,
"Streaming"
,
"Stratosphere"
,
"Streaming"
,
"Stratosphere"
,
"Streaming"
,
"Streaming"
));
t
=
System
.
nanoTime
();
for
(
int
i
=
0
;
i
<
ITERATION
;
i
++)
{
...
...
@@ -312,19 +304,16 @@ public class StreamRecordTest {
int
num
=
42
;
String
str
=
"above clouds"
;
StreamRecord
rec
=
new
StreamRecord
(
new
Tuple2
<
Integer
,
String
>(
num
,
str
));
StreamRecord
rec
=
new
StreamRecord
(
new
Tuple2
<
Integer
,
String
>(
num
,
str
));
try
{
rec
.
write
(
out
);
DataInputStream
in
=
new
DataInputStream
(
new
ByteArrayInputStream
(
buff
.
toByteArray
()));
DataInputStream
in
=
new
DataInputStream
(
new
ByteArrayInputStream
(
buff
.
toByteArray
()));
StreamRecord
newRec
=
new
StreamRecord
();
newRec
.
read
(
in
);
@SuppressWarnings
(
"unchecked"
)
Tuple2
<
Integer
,
String
>
tupleOut
=
(
Tuple2
<
Integer
,
String
>)
newRec
.
getTuple
(
0
);
Tuple2
<
Integer
,
String
>
tupleOut
=
(
Tuple2
<
Integer
,
String
>)
newRec
.
getTuple
(
0
);
assertEquals
(
tupleOut
.
getField
(
0
),
42
);
}
catch
(
IOException
e
)
{
...
...
@@ -353,4 +342,64 @@ public class StreamRecordTest {
}
@Test
public
void
typeCopyTest
()
throws
NoSuchTupleException
,
IOException
{
StreamRecord
rec
=
new
StreamRecord
(
new
Tuple9
<
Boolean
,
Byte
,
Character
,
Double
,
Float
,
Integer
,
Long
,
Short
,
String
>((
Boolean
)
true
,
(
Byte
)
(
byte
)
12
,
(
Character
)
'a'
,
(
Double
)
12.5
,
(
Float
)
(
float
)
13.5
,
(
Integer
)
1234
,
(
Long
)
12345678900
l
,
(
Short
)
(
short
)
12345
,
"something"
));
@SuppressWarnings
(
"rawtypes"
)
Class
[]
types
=
new
Class
[
9
];
assertEquals
(
new
Class
[]
{
Boolean
.
class
,
Byte
.
class
,
Character
.
class
,
Double
.
class
,
Float
.
class
,
Integer
.
class
,
Long
.
class
,
Short
.
class
,
String
.
class
},
rec
.
tupleBasicTypesFromLong
(
rec
.
tupleBasicTypesToLong
(
rec
.
getTuple
()),
9
));
assertEquals
(
new
Class
[]
{
Boolean
.
class
,
Byte
.
class
,
Character
.
class
,
Double
.
class
,
Float
.
class
,
Integer
.
class
,
Long
.
class
,
Short
.
class
,
String
.
class
},
rec
.
tupleBasicTypesFromString
(
rec
.
tupleBasicTypesToString
(
rec
.
getTuple
()),
9
));
assertEquals
(
new
Class
[]
{
Boolean
.
class
,
Byte
.
class
,
Character
.
class
,
Double
.
class
,
Float
.
class
,
Integer
.
class
,
Long
.
class
,
Short
.
class
,
String
.
class
},
rec
.
tupleBasicTypesFromByteArray
(
rec
.
tupleBasicTypesToByteArray
(
rec
.
getTuple
()),
9
));
ByteArrayOutputStream
buff
=
new
ByteArrayOutputStream
();
DataOutputStream
out
=
new
DataOutputStream
(
buff
);
long
start
=
System
.
nanoTime
();
for
(
int
i
=
0
;
i
<
1000
;
i
++)
{
out
.
writeLong
(
rec
.
tupleBasicTypesToLong
(
rec
.
getTuple
()));
}
DataInputStream
in
=
new
DataInputStream
(
new
ByteArrayInputStream
(
buff
.
toByteArray
()));
for
(
int
i
=
0
;
i
<
1000
;
i
++)
{
rec
.
tupleBasicTypesFromLong
(
in
.
readLong
(),
9
);
}
System
.
out
.
println
(
"Type copy with Long:\t\t"
+
(
System
.
nanoTime
()
-
start
)
+
" ns"
);
ByteArrayOutputStream
buff2
=
new
ByteArrayOutputStream
();
DataOutputStream
out2
=
new
DataOutputStream
(
buff2
);
start
=
System
.
nanoTime
();
for
(
int
i
=
0
;
i
<
1000
;
i
++)
{
StringValue
sv
=
new
StringValue
(
rec
.
tupleBasicTypesToString
(
rec
.
getTuple
()));
sv
.
write
(
out2
);
}
DataInputStream
in2
=
new
DataInputStream
(
new
ByteArrayInputStream
(
buff2
.
toByteArray
()));
for
(
int
i
=
0
;
i
<
1000
;
i
++)
{
StringValue
sv
=
new
StringValue
(
""
);
sv
.
read
(
in2
);
Class
[]
types2
=
rec
.
tupleBasicTypesFromString
(
sv
.
getValue
(),
9
);
}
System
.
out
.
println
(
"Type copy with String:\t\t"
+
(
System
.
nanoTime
()
-
start
)
+
" ns"
);
ByteArrayOutputStream
buff3
=
new
ByteArrayOutputStream
();
DataOutputStream
out3
=
new
DataOutputStream
(
buff3
);
start
=
System
.
nanoTime
();
for
(
int
i
=
0
;
i
<
1000
;
i
++)
{
out3
.
write
(
rec
.
tupleBasicTypesToByteArray
(
rec
.
getTuple
()));
}
DataInputStream
in3
=
new
DataInputStream
(
new
ByteArrayInputStream
(
buff3
.
toByteArray
()));
for
(
int
i
=
0
;
i
<
1000
;
i
++)
{
byte
[]
byteTypes
=
new
byte
[
9
];
in3
.
read
(
byteTypes
);
Class
[]
types2
=
rec
.
tupleBasicTypesFromByteArray
(
byteTypes
,
9
);
}
System
.
out
.
println
(
"Type copy with ByteArray:\t"
+
(
System
.
nanoTime
()
-
start
)
+
" ns"
);
}
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录