Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
doujutun3207
flink
提交
6c261d3a
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,发现更多精彩内容 >>
提交
6c261d3a
编写于
3月 03, 2014
作者:
S
StephanEwen
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
CsvInputFormat can optionally skip an initial header line.
上级
6344e573
变更
4
隐藏空白更改
内联
并排
Showing
4 changed file
with
179 addition
and
63 deletion
+179
-63
stratosphere-core/src/main/java/eu/stratosphere/api/common/io/DelimitedInputFormat.java
...a/eu/stratosphere/api/common/io/DelimitedInputFormat.java
+1
-1
stratosphere-core/src/main/java/eu/stratosphere/api/common/io/GenericCsvInputFormat.java
.../eu/stratosphere/api/common/io/GenericCsvInputFormat.java
+45
-29
stratosphere-core/src/test/java/eu/stratosphere/api/common/io/FileOutputFormatTest.java
...a/eu/stratosphere/api/common/io/FileOutputFormatTest.java
+1
-3
stratosphere-core/src/test/java/eu/stratosphere/api/common/io/GenericCsvInputFormatTest.java
...stratosphere/api/common/io/GenericCsvInputFormatTest.java
+132
-30
未找到文件。
stratosphere-core/src/main/java/eu/stratosphere/api/common/io/DelimitedInputFormat.java
浏览文件 @
6c261d3a
...
...
@@ -467,7 +467,7 @@ public abstract class DelimitedInputFormat<OT> extends FileInputFormat<OT> {
// --------------------------------------------------------------------------------------------
pr
ivate
boolean
readLine
()
throws
IOException
{
pr
otected
final
boolean
readLine
()
throws
IOException
{
if
(
this
.
stream
==
null
||
this
.
overLimit
)
{
return
false
;
}
...
...
stratosphere-core/src/main/java/eu/stratosphere/api/common/io/GenericCsvInputFormat.java
浏览文件 @
6c261d3a
...
...
@@ -55,7 +55,9 @@ public abstract class GenericCsvInputFormat<OT> extends DelimitedInputFormat<OT>
private
char
fieldDelim
=
DEFAULT_FIELD_DELIMITER
;
private
boolean
lenient
=
false
;
private
boolean
lenient
;
private
boolean
skipFirstLineAsHeader
;
// --------------------------------------------------------------------------------------------
...
...
@@ -131,38 +133,39 @@ public abstract class GenericCsvInputFormat<OT> extends DelimitedInputFormat<OT>
this
.
fieldTypes
=
denseTypeArray
;
}
public
void
setFields
(
int
[]
sourceFieldIndices
,
Class
<?
extends
Value
>[]
fieldTypes
)
{
Preconditions
.
checkNotNull
(
fieldTypes
);
Preconditions
.
checkArgument
(
sourceFieldIndices
.
length
==
fieldTypes
.
length
,
"Number of field indices and field types must match."
);
public
void
setFields
(
int
[]
sourceFieldIndices
,
Class
<?
extends
Value
>[]
fieldTypes
)
{
Preconditions
.
checkNotNull
(
fieldTypes
);
Preconditions
.
checkArgument
(
sourceFieldIndices
.
length
==
fieldTypes
.
length
,
"Number of field indices and field types must match."
);
for
(
int
i
:
sourceFieldIndices
)
{
if
(
i
<
0
)
{
throw
new
IllegalArgumentException
(
"Field indices must not be smaller than zero."
);
}
}
for
(
int
i
:
sourceFieldIndices
)
{
if
(
i
<
0
)
{
throw
new
IllegalArgumentException
(
"Field indices must not be smaller than zero."
);
}
}
int
largestFieldIndex
=
Ints
.
max
(
sourceFieldIndices
);
this
.
fieldIncluded
=
new
boolean
[
largestFieldIndex
+
1
];
ArrayList
<
Class
<?
extends
Value
>>
types
=
new
ArrayList
<
Class
<?
extends
Value
>>();
int
largestFieldIndex
=
Ints
.
max
(
sourceFieldIndices
);
this
.
fieldIncluded
=
new
boolean
[
largestFieldIndex
+
1
];
ArrayList
<
Class
<?
extends
Value
>>
types
=
new
ArrayList
<
Class
<?
extends
Value
>>();
// check if we support parsers for these types
for
(
int
i
=
0
;
i
<
fieldTypes
.
length
;
i
++)
{
Class
<?
extends
Value
>
type
=
fieldTypes
[
i
];
// check if we support parsers for these types
for
(
int
i
=
0
;
i
<
fieldTypes
.
length
;
i
++)
{
Class
<?
extends
Value
>
type
=
fieldTypes
[
i
];
if
(
type
!=
null
)
{
if
(
FieldParser
.
getParserForType
(
type
)
==
null
)
{
throw
new
IllegalArgumentException
(
"The type '"
+
type
.
getName
()
+
"' is not supported for the CSV input format."
);
}
types
.
add
(
type
);
fieldIncluded
[
sourceFieldIndices
[
i
]]
=
true
;
}
}
if
(
type
!=
null
)
{
if
(
FieldParser
.
getParserForType
(
type
)
==
null
)
{
throw
new
IllegalArgumentException
(
"The type '"
+
type
.
getName
()
+
"' is not supported for the CSV input format."
);
}
types
.
add
(
type
);
fieldIncluded
[
sourceFieldIndices
[
i
]]
=
true
;
}
}
@SuppressWarnings
(
"unchecked"
)
Class
<?
extends
Value
>[]
denseTypeArray
=
(
Class
<?
extends
Value
>[])
types
.
toArray
(
new
Class
[
types
.
size
()]);
this
.
fieldTypes
=
denseTypeArray
;
}
@SuppressWarnings
(
"unchecked"
)
Class
<?
extends
Value
>[]
denseTypeArray
=
(
Class
<?
extends
Value
>[])
types
.
toArray
(
new
Class
[
types
.
size
()]);
this
.
fieldTypes
=
denseTypeArray
;
}
public
int
getNumberOfFieldsTotal
()
{
return
this
.
fieldIncluded
.
length
;
...
...
@@ -191,6 +194,14 @@ public abstract class GenericCsvInputFormat<OT> extends DelimitedInputFormat<OT>
this
.
lenient
=
lenient
;
}
public
boolean
isSkippingFirstLineAsHeader
()
{
return
skipFirstLineAsHeader
;
}
public
void
setSkipFirstLineAsHeader
(
boolean
skipFirstLine
)
{
this
.
skipFirstLineAsHeader
=
skipFirstLine
;
}
protected
FieldParser
<
Value
>[]
getFieldParsers
()
{
return
this
.
fieldParsers
;
}
...
...
@@ -220,6 +231,11 @@ public abstract class GenericCsvInputFormat<OT> extends DelimitedInputFormat<OT>
}
}
this
.
fieldParsers
=
parsers
;
// skip the first line, if we are at the beginning of a file and have the option set
if
(
this
.
skipFirstLineAsHeader
&&
this
.
splitStart
==
0
)
{
readLine
();
// read and ignore
}
}
protected
boolean
parseRecord
(
Value
[]
valueHolders
,
byte
[]
bytes
,
int
offset
,
int
numBytes
)
throws
ParseException
{
...
...
@@ -270,7 +286,7 @@ public abstract class GenericCsvInputFormat<OT> extends DelimitedInputFormat<OT>
field
+=
(
skipCnt
-
1
);
}
else
if
(
lenient
)
{
// no valid line, but we
go on
// no valid line, but we
do not report an exception, simply skip the line
return
false
;
}
else
{
...
...
stratosphere-core/src/test/java/eu/stratosphere/api/common/io/FileOutputFormatTest.java
浏览文件 @
6c261d3a
...
...
@@ -5,20 +5,18 @@ import java.io.IOException;
import
junit.framework.Assert
;
import
org.apache.log4j.Level
;
import
org.junit.BeforeClass
;
import
org.junit.Test
;
import
eu.stratosphere.configuration.Configuration
;
import
eu.stratosphere.types.IntValue
;
import
eu.stratosphere.util.LogUtils
;
import
eu.stratosphere.util.OperatingSystem
;
public
class
FileOutputFormatTest
{
@BeforeClass
public
static
void
initialize
()
{
LogUtils
.
initializeDefault
ConsoleLogger
(
Level
.
WARN
);
LogUtils
.
initializeDefault
TestConsoleLogger
(
);
}
@Test
...
...
stratosphere-core/src/test/java/eu/stratosphere/api/common/io/GenericCsvInputFormatTest.java
浏览文件 @
6c261d3a
...
...
@@ -196,42 +196,41 @@ public class GenericCsvInputFormatTest {
}
}
@SuppressWarnings
(
"unchecked"
)
@Test
public
void
testSparseParseWithIndices
()
{
try
{
final
String
fileContent
=
"111|222|333|444|555|666|777|888|999|000|\n000|999|888|777|666|555|444|333|222|111|"
;
final
FileInputSplit
split
=
createTempFile
(
fileContent
);
@SuppressWarnings
(
"unchecked"
)
@Test
public
void
testSparseParseWithIndices
()
{
try
{
final
String
fileContent
=
"111|222|333|444|555|666|777|888|999|000|\n000|999|888|777|666|555|444|333|222|111|"
;
final
FileInputSplit
split
=
createTempFile
(
fileContent
);
final
Configuration
parameters
=
new
Configuration
();
final
Configuration
parameters
=
new
Configuration
();
format
.
setFieldDelim
(
'|'
);
format
.
setFields
(
new
int
[]{
0
,
3
,
7
},
(
Class
<?
extends
Value
>[])
new
Class
[]{
IntValue
.
class
,
IntValue
.
class
,
IntValue
.
class
});
format
.
configure
(
parameters
);
format
.
open
(
split
);
format
.
setFieldDelim
(
'|'
);
format
.
setFields
(
new
int
[]
{
0
,
3
,
7
},
(
Class
<?
extends
Value
>[])
new
Class
[]
{
IntValue
.
class
,
IntValue
.
class
,
IntValue
.
class
});
format
.
configure
(
parameters
);
format
.
open
(
split
);
Value
[]
values
=
createIntValues
(
3
);
Value
[]
values
=
createIntValues
(
3
);
assertTrue
(
format
.
nextRecord
(
values
));
assertEquals
(
111
,
((
IntValue
)
values
[
0
]).
getValue
());
assertEquals
(
444
,
((
IntValue
)
values
[
1
]).
getValue
());
assertEquals
(
888
,
((
IntValue
)
values
[
2
]).
getValue
());
assertTrue
(
format
.
nextRecord
(
values
));
assertEquals
(
111
,
((
IntValue
)
values
[
0
]).
getValue
());
assertEquals
(
444
,
((
IntValue
)
values
[
1
]).
getValue
());
assertEquals
(
888
,
((
IntValue
)
values
[
2
]).
getValue
());
assertTrue
(
format
.
nextRecord
(
values
));
assertEquals
(
000
,
((
IntValue
)
values
[
0
]).
getValue
());
assertEquals
(
777
,
((
IntValue
)
values
[
1
]).
getValue
());
assertEquals
(
333
,
((
IntValue
)
values
[
2
]).
getValue
());
assertTrue
(
format
.
nextRecord
(
values
));
assertEquals
(
000
,
((
IntValue
)
values
[
0
]).
getValue
());
assertEquals
(
777
,
((
IntValue
)
values
[
1
]).
getValue
());
assertEquals
(
333
,
((
IntValue
)
values
[
2
]).
getValue
());
assertFalse
(
format
.
nextRecord
(
values
));
assertTrue
(
format
.
reachedEnd
());
}
catch
(
Exception
ex
)
{
System
.
err
.
println
(
ex
.
getMessage
());
ex
.
printStackTrace
();
Assert
.
fail
(
"Test erroneous"
);
}
}
assertFalse
(
format
.
nextRecord
(
values
));
assertTrue
(
format
.
reachedEnd
());
}
catch
(
Exception
ex
)
{
System
.
err
.
println
(
ex
.
getMessage
());
ex
.
printStackTrace
();
Assert
.
fail
(
"Test erroneous"
);
}
}
@SuppressWarnings
(
"unchecked"
)
@Test
...
...
@@ -346,6 +345,37 @@ public class GenericCsvInputFormatTest {
}
}
@SuppressWarnings
(
"unchecked"
)
@Test
public
void
testReadInvalidContentsLenientWithSkipping
()
{
try
{
final
String
fileContent
=
"abc|dfgsdf|777|444\n"
+
// good line
"kkz|777|foobar|hhg\n"
+
// wrong data type in field
"kkz|777foobarhhg \n"
+
// too short, a skipped field never ends
"xyx|ignored|42|\n"
;
// another good line
final
FileInputSplit
split
=
createTempFile
(
fileContent
);
final
Configuration
parameters
=
new
Configuration
();
format
.
setFieldDelim
(
'|'
);
format
.
setFieldTypes
(
StringValue
.
class
,
null
,
IntValue
.
class
);
format
.
setLenient
(
true
);
format
.
configure
(
parameters
);
format
.
open
(
split
);
Value
[]
values
=
new
Value
[]
{
new
StringValue
(),
new
IntValue
()};
assertTrue
(
format
.
nextRecord
(
values
));
assertFalse
(
format
.
nextRecord
(
values
));
assertFalse
(
format
.
nextRecord
(
values
));
assertTrue
(
format
.
nextRecord
(
values
));
}
catch
(
Exception
ex
)
{
Assert
.
fail
(
"Test failed due to a "
+
ex
.
getClass
().
getSimpleName
()
+
": "
+
ex
.
getMessage
());
}
}
@SuppressWarnings
(
"unchecked"
)
@Test
public
void
readWithEmptyField
()
{
...
...
@@ -383,6 +413,78 @@ public class GenericCsvInputFormatTest {
Assert
.
fail
(
"Test failed due to a "
+
ex
.
getClass
().
getSimpleName
()
+
": "
+
ex
.
getMessage
());
}
}
@SuppressWarnings
(
"unchecked"
)
@Test
public
void
readWithHeaderLine
()
{
try
{
final
String
fileContent
=
"colname-1|colname-2|some name 3|column four|\n"
+
"123|abc|456|def|\n"
+
"987|xyz|654|pqr|\n"
;
final
FileInputSplit
split
=
createTempFile
(
fileContent
);
final
Configuration
parameters
=
new
Configuration
();
format
.
setFieldDelim
(
'|'
);
format
.
setFieldTypes
(
IntValue
.
class
,
StringValue
.
class
,
IntValue
.
class
,
StringValue
.
class
);
format
.
setSkipFirstLineAsHeader
(
true
);
format
.
configure
(
parameters
);
format
.
open
(
split
);
Value
[]
values
=
new
Value
[]
{
new
IntValue
(),
new
StringValue
(),
new
IntValue
(),
new
StringValue
()};
// first line is skipped as header
assertTrue
(
format
.
nextRecord
(
values
));
// first row (= second line)
assertTrue
(
format
.
nextRecord
(
values
));
// second row (= third line)
assertFalse
(
format
.
nextRecord
(
values
));
// exhausted
assertTrue
(
format
.
reachedEnd
());
// exhausted
}
catch
(
Exception
ex
)
{
Assert
.
fail
(
"Test failed due to a "
+
ex
.
getClass
().
getSimpleName
()
+
": "
+
ex
.
getMessage
());
}
}
@SuppressWarnings
(
"unchecked"
)
@Test
public
void
readWithHeaderLineAndInvalidIntermediate
()
{
try
{
final
String
fileContent
=
"colname-1|colname-2|some name 3|column four|\n"
+
"123|abc|456|def|\n"
+
"colname-1|colname-2|some name 3|column four|\n"
+
// repeated header in the middle
"987|xyz|654|pqr|\n"
;
final
FileInputSplit
split
=
createTempFile
(
fileContent
);
final
Configuration
parameters
=
new
Configuration
();
format
.
setFieldDelim
(
'|'
);
format
.
setFieldTypes
(
IntValue
.
class
,
StringValue
.
class
,
IntValue
.
class
,
StringValue
.
class
);
format
.
setSkipFirstLineAsHeader
(
true
);
format
.
configure
(
parameters
);
format
.
open
(
split
);
Value
[]
values
=
new
Value
[]
{
new
IntValue
(),
new
StringValue
(),
new
IntValue
(),
new
StringValue
()};
// first line is skipped as header
assertTrue
(
format
.
nextRecord
(
values
));
// first row (= second line)
try
{
format
.
nextRecord
(
values
);
Assert
.
fail
(
"Format accepted invalid line."
);
}
catch
(
ParseException
e
)
{
// as we expected
}
}
catch
(
Exception
ex
)
{
Assert
.
fail
(
"Test failed due to a "
+
ex
.
getClass
().
getSimpleName
()
+
": "
+
ex
.
getMessage
());
}
}
private
FileInputSplit
createTempFile
(
String
content
)
throws
IOException
{
this
.
tempFile
=
File
.
createTempFile
(
"test_contents"
,
"tmp"
);
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录