You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Flavio Pompermaier <po...@okkam.it> on 2017/09/08 14:27:36 UTC
BucketingSink never closed
Hi to all,
I'm trying to test a streaming job but the files written by
the BucketingSink are never finalized (remains into the pending state).
Is this caused by the fact that the job finishes before the checkpoint?
Shouldn't the sink properly close anyway?
This is my code:
@Test
public void testBucketingSink() throws Exception {
final StreamExecutionEnvironment senv =
StreamExecutionEnvironment.getExecutionEnvironment();
final StreamTableEnvironment tEnv =
TableEnvironment.getTableEnvironment(senv);
senv.enableCheckpointing(5000);
DataStream<String> testStream = senv.fromElements(//
"1,aaa,white", //
"2,bbb,gray", //
"3,ccc,white", //
"4,bbb,gray", //
"5,bbb,gray" //
);
final RowTypeInfo rtf = new RowTypeInfo(
BasicTypeInfo.STRING_TYPE_INFO,
BasicTypeInfo.STRING_TYPE_INFO,
BasicTypeInfo.STRING_TYPE_INFO);
DataStream<Row> rows = testStream.map(new MapFunction<String, Row>() {
private static final long serialVersionUID = 1L;
@Override
public Row map(String str) throws Exception {
String[] split = str.split(Pattern.quote(","));
Row ret = new Row(3);
ret.setField(0, split[0]);
ret.setField(1, split[1]);
ret.setField(2, split[2]);
return ret;
}
}).returns(rtf);
String columnNames = "id,value,state";
final String dsName = "test";
tEnv.registerDataStream(dsName, rows, columnNames);
final String whiteAreaFilter = "state = 'white'";
DataStream<Row> grayArea = rows;
DataStream<Row> whiteArea = null;
if (whiteAreaFilter != null) {
String sql = "SELECT *, (%s) as _WHITE FROM %s";
sql = String.format(sql, whiteAreaFilter, dsName);
Table table = tEnv.sql(sql);
grayArea =
tEnv.toDataStream(table.where("!_WHITE").select(columnNames), rtf);
DataStream<Row> nw =
tEnv.toDataStream(table.where("_WHITE").select(columnNames), rtf);
whiteArea = whiteArea == null ? nw : whiteArea.union(nw);
}
Writer<Row> bucketSinkwriter = new RowCsvWriter("UTF-8", "\t", "\n");
String datasetWhiteDir = "/tmp/bucket/white";
BucketingSink<Row> whiteAreaSink = new
BucketingSink<>(datasetWhiteDir.toString());
whiteAreaSink.setWriter(bucketSinkwriter);
whiteAreaSink.setBatchSize(10);
whiteArea.addSink(whiteAreaSink);
String datasetGrayDir = "/tmp/bucket/gray";
BucketingSink<Row> grayAreaSink = new
BucketingSink<>(datasetGrayDir.toString());
grayAreaSink.setWriter(bucketSinkwriter);
grayAreaSink.setBatchSize(10);
grayArea.addSink(grayAreaSink);
JobExecutionResult jobInfo = senv.execute("Buketing sink test ");
System.out.printf("Job took %s minutes",
jobInfo.getNetRuntime(TimeUnit.MINUTES));
}
public class RowCsvWriter extends StreamWriterBase<Row> {
private static final long serialVersionUID = 1L;
private final String charsetName;
private transient Charset charset;
private String fieldDelimiter;
private String recordDelimiter;
private boolean allowNullValues = true;
private boolean quoteStrings = false;
/**
* Creates a new {@code StringWriter} that uses {@code "UTF-8"} charset
to convert strings to
* bytes.
*/
public RowCsvWriter() {
this("UTF-8", CsvOutputFormat.DEFAULT_FIELD_DELIMITER,
CsvOutputFormat.DEFAULT_LINE_DELIMITER);
}
/**
* Creates a new {@code StringWriter} that uses the given charset to
convert strings to bytes.
*
* @param charsetName Name of the charset to be used, must be valid input
for
* {@code Charset.forName(charsetName)}
*/
public RowCsvWriter(String charsetName, String fieldDelimiter, String
recordDelimiter) {
this.charsetName = charsetName;
this.fieldDelimiter = fieldDelimiter;
this.recordDelimiter = recordDelimiter;
}
@Override
public void open(FileSystem fs, Path path) throws IOException {
super.open(fs, path);
try {
this.charset = Charset.forName(charsetName);
} catch (IllegalCharsetNameException ex) {
throw new IOException("The charset " + charsetName + " is not
valid.", ex);
} catch (UnsupportedCharsetException ex) {
throw new IOException("The charset " + charsetName + " is not
supported.", ex);
}
}
@Override
public void write(Row element) throws IOException {
FSDataOutputStream outputStream = getStream();
writeRow(element, outputStream);
}
private void writeRow(Row element, FSDataOutputStream out) throws
IOException {
int numFields = element.getArity();
for (int i = 0; i < numFields; i++) {
Object obj = element.getField(i);
if (obj != null) {
if (i != 0) {
out.write(this.fieldDelimiter.getBytes(charset));
}
if (quoteStrings) {
if (obj instanceof String || obj instanceof StringValue) {
out.write('"');
out.write(obj.toString().getBytes(charset));
out.write('"');
} else {
out.write(obj.toString().getBytes(charset));
}
} else {
out.write(obj.toString().getBytes(charset));
}
} else {
if (this.allowNullValues) {
if (i != 0) {
out.write(this.fieldDelimiter.getBytes(charset));
}
} else {
throw new RuntimeException("Cannot write tuple with <null> value
at position: " + i);
}
}
}
// add the record delimiter
out.write(this.recordDelimiter.getBytes(charset));
}
@Override
public Writer<Row> duplicate() {
return new RowCsvWriter(charsetName, fieldDelimiter, recordDelimiter);
}
}
Any help is appreciated,
Flavio
Re: BucketingSink never closed
Posted by "Tzu-Li (Gordon) Tai" <tz...@apache.org>.
Ah, sorry, one correction. Just realized there’s already some analysis of the BucketingSink closing issue in this mail thread.
Please ignore my request for relevant logs :)
On 13 September 2017 at 10:56:10 AM, Tzu-Li (Gordon) Tai (tzulitai@apache.org) wrote:
Hi Flavio,
Let me try to understand / look at some of the problems you have encountered.
checkpointing: it's not clear which checkpointing system to use and how to tune/monitor it and avoid OOM exceptions.
What do you mean be which "checkpointing system” to use? Do you mean state backends? Typically, you would only get OOM exceptions for memory-backed state backends if the state size exceeds the memory capacity. State sizes can be queried from the REST APIs / Web UI.
cleanup: BucketingSink doesn't always move to final state
This sounds like a bug that we should look into. Do you have any logs on which you observed this?
missing output formats: parquet support to write generic Rows not very well supported (at least out of the box) [1]
Would you be up to opening up JIRAs for what you think is missing (if there isn’t one already)?
progress monitoring: for example in the ES connector there's no way (apart from using accumulators) to monitor the progress of the indexing
Maybe we can add some built-in metric in the ES sink connector that tracks the number of successfully indexed elements, which can then be queried from the REST API / Web UI. That wouldn’t be too much effort. What do you think, would that be useful for your case?
Would be happy to hear your thoughts on this!
Cheers,
Gordon
On 12 September 2017 at 11:36:27 AM, Flavio Pompermaier (pompermaier@okkam.it) wrote:
For the moment I give up with streaming...too many missing/unclear features wrt batch.
For example:
checkpointing: it's not clear which checkpointing system to use and how to tune/monitor it and avoid OOM exceptions. Moreover is it really necessary to use it? For example if I read a file from HDFS and I don't have a checkpoint it could be ok to re-run the job on all the data in case of errors (i.e. the stream is managed like a batch)
cleanup: BucketingSink doesn't always move to final state
missing output formats: parquet support to write generic Rows not very well supported (at least out of the box) [1]
progress monitoring: for example in the ES connector there's no way (apart from using accumulators) to monitor the progress of the indexing
[1] https://stackoverflow.com/questions/41144659/flink-avro-parquet-writer-in-rollingsink
Maybe I'm wrong with those points but the attempt to replace my current batch system with a streaming one had no luck with those points.
Best,
Flavio
On Fri, Sep 8, 2017 at 5:29 PM, Aljoscha Krettek <al...@apache.org> wrote:
Hi,
Expanding a bit on Kostas' answer. Yes, your analysis is correct, the problem is that the job is shutting down before a last checkpoint can "confirm" the written bucket data by moving it to the final state. The problem, as Kostas noted is that a user function (and thus also BucketingSink) does not know whether close() is being called because of a failure or because normal job shutdown. Therefore, we cannot move data to the final stage there.
Once we have the issue that Kostas posted resolve we can also resolve this problem for the BucketingSink.
Best,
Aljoscha
On 8. Sep 2017, at 16:48, Kostas Kloudas <k....@data-artisans.com> wrote:
Hi Flavio,
If I understand correctly, I think you bumped into this issue: https://issues.apache.org/jira/browse/FLINK-2646
There is also a similar discussion on the BucketingSink here:
http://apache-flink-mailing-list-archive.1008284.n3.nabble.com/DISCUSS-Adding-a-dispose-method-in-the-RichFunction-td14466.html#a14468
Kostas
On Sep 8, 2017, at 4:27 PM, Flavio Pompermaier <po...@okkam.it> wrote:
Hi to all,
I'm trying to test a streaming job but the files written by the BucketingSink are never finalized (remains into the pending state).
Is this caused by the fact that the job finishes before the checkpoint?
Shouldn't the sink properly close anyway?
This is my code:
@Test
public void testBucketingSink() throws Exception {
final StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
final StreamTableEnvironment tEnv = TableEnvironment.getTableEnvironment(senv);
senv.enableCheckpointing(5000);
DataStream<String> testStream = senv.fromElements(//
"1,aaa,white", //
"2,bbb,gray", //
"3,ccc,white", //
"4,bbb,gray", //
"5,bbb,gray" //
);
final RowTypeInfo rtf = new RowTypeInfo(
BasicTypeInfo.STRING_TYPE_INFO,
BasicTypeInfo.STRING_TYPE_INFO,
BasicTypeInfo.STRING_TYPE_INFO);
DataStream<Row> rows = testStream.map(new MapFunction<String, Row>() {
private static final long serialVersionUID = 1L;
@Override
public Row map(String str) throws Exception {
String[] split = str.split(Pattern.quote(","));
Row ret = new Row(3);
ret.setField(0, split[0]);
ret.setField(1, split[1]);
ret.setField(2, split[2]);
return ret;
}
}).returns(rtf);
String columnNames = "id,value,state";
final String dsName = "test";
tEnv.registerDataStream(dsName, rows, columnNames);
final String whiteAreaFilter = "state = 'white'";
DataStream<Row> grayArea = rows;
DataStream<Row> whiteArea = null;
if (whiteAreaFilter != null) {
String sql = "SELECT *, (%s) as _WHITE FROM %s";
sql = String.format(sql, whiteAreaFilter, dsName);
Table table = tEnv.sql(sql);
grayArea = tEnv.toDataStream(table.where("!_WHITE").select(columnNames), rtf);
DataStream<Row> nw = tEnv.toDataStream(table.where("_WHITE").select(columnNames), rtf);
whiteArea = whiteArea == null ? nw : whiteArea.union(nw);
}
Writer<Row> bucketSinkwriter = new RowCsvWriter("UTF-8", "\t", "\n");
String datasetWhiteDir = "/tmp/bucket/white";
BucketingSink<Row> whiteAreaSink = new BucketingSink<>(datasetWhiteDir.toString());
whiteAreaSink.setWriter(bucketSinkwriter);
whiteAreaSink.setBatchSize(10);
whiteArea.addSink(whiteAreaSink);
String datasetGrayDir = "/tmp/bucket/gray";
BucketingSink<Row> grayAreaSink = new BucketingSink<>(datasetGrayDir.toString());
grayAreaSink.setWriter(bucketSinkwriter);
grayAreaSink.setBatchSize(10);
grayArea.addSink(grayAreaSink);
JobExecutionResult jobInfo = senv.execute("Buketing sink test ");
System.out.printf("Job took %s minutes", jobInfo.getNetRuntime(TimeUnit.MINUTES));
}
public class RowCsvWriter extends StreamWriterBase<Row> {
private static final long serialVersionUID = 1L;
private final String charsetName;
private transient Charset charset;
private String fieldDelimiter;
private String recordDelimiter;
private boolean allowNullValues = true;
private boolean quoteStrings = false;
/**
* Creates a new {@code StringWriter} that uses {@code "UTF-8"} charset to convert strings to
* bytes.
*/
public RowCsvWriter() {
this("UTF-8", CsvOutputFormat.DEFAULT_FIELD_DELIMITER, CsvOutputFormat.DEFAULT_LINE_DELIMITER);
}
/**
* Creates a new {@code StringWriter} that uses the given charset to convert strings to bytes.
*
* @param charsetName Name of the charset to be used, must be valid input for
* {@code Charset.forName(charsetName)}
*/
public RowCsvWriter(String charsetName, String fieldDelimiter, String recordDelimiter) {
this.charsetName = charsetName;
this.fieldDelimiter = fieldDelimiter;
this.recordDelimiter = recordDelimiter;
}
@Override
public void open(FileSystem fs, Path path) throws IOException {
super.open(fs, path);
try {
this.charset = Charset.forName(charsetName);
} catch (IllegalCharsetNameException ex) {
throw new IOException("The charset " + charsetName + " is not valid.", ex);
} catch (UnsupportedCharsetException ex) {
throw new IOException("The charset " + charsetName + " is not supported.", ex);
}
}
@Override
public void write(Row element) throws IOException {
FSDataOutputStream outputStream = getStream();
writeRow(element, outputStream);
}
private void writeRow(Row element, FSDataOutputStream out) throws IOException {
int numFields = element.getArity();
for (int i = 0; i < numFields; i++) {
Object obj = element.getField(i);
if (obj != null) {
if (i != 0) {
out.write(this.fieldDelimiter.getBytes(charset));
}
if (quoteStrings) {
if (obj instanceof String || obj instanceof StringValue) {
out.write('"');
out.write(obj.toString().getBytes(charset));
out.write('"');
} else {
out.write(obj.toString().getBytes(charset));
}
} else {
out.write(obj.toString().getBytes(charset));
}
} else {
if (this.allowNullValues) {
if (i != 0) {
out.write(this.fieldDelimiter.getBytes(charset));
}
} else {
throw new RuntimeException("Cannot write tuple with <null> value at position: " + i);
}
}
}
// add the record delimiter
out.write(this.recordDelimiter.getBytes(charset));
}
@Override
public Writer<Row> duplicate() {
return new RowCsvWriter(charsetName, fieldDelimiter, recordDelimiter);
}
}
Any help is appreciated,
Flavio
Re: BucketingSink never closed
Posted by Flavio Pompermaier <po...@okkam.it>.
I've just looked at Robert presentation at FF [1] and that's exactly what I
was waiting for streaming planning/training...
Very useful ;)
[1] https://www.youtube.com/watch?v=8l8dCKMMWkw
On Wed, Sep 13, 2017 at 12:04 PM, Flavio Pompermaier <po...@okkam.it>
wrote:
> Hi Gordon,
> thanks for your feedback. The main problem for me is that moving from
> batch to stream should be much easier IMHO.
>
> Rows should be a first class citizen in Flink and should be VERY easy to
> read/write them, while at the moment it seems that Tuples are the
> dominating type...I don't want to write a serializer/outputFormat to
> persist Rows as Parquet, Avro, Thrift, OCR, Kudu, Hive, etc..I expect to
> have some already existing (and mantained) connector already available
> somewhere. The case of the Parquet Rollink sink is just an example.
>
> Regarding state backends I think that its not so easy to understand how to
> design and monitor it properly: there are many parameters/variables to take
> into account and it would be helpful to have a proper hands-on training
> course/certification about this...
>
> About ES indexing monitoring see my discussion with Chesnay at
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/
> Streaming-job-monitoring-td13583.html: what I need is just to have
> recordsIn/recordsOut reflecting real values.
>
> Best,
> Flavio
>
> On Wed, Sep 13, 2017 at 10:56 AM, Tzu-Li (Gordon) Tai <tzulitai@apache.org
> > wrote:
>
>> Hi Flavio,
>>
>> Let me try to understand / look at some of the problems you have
>> encountered.
>>
>>
>> - checkpointing: it's not clear which checkpointing system to use and
>> how to tune/monitor it and avoid OOM exceptions.
>>
>> What do you mean be which "checkpointing system” to use? Do you mean
>> state backends? Typically, you would only get OOM exceptions for
>> memory-backed state backends if the state size exceeds the memory capacity.
>> State sizes can be queried from the REST APIs / Web UI.
>>
>>
>> - cleanup: BucketingSink doesn't always move to final state
>>
>> This sounds like a bug that we should look into. Do you have any logs on
>> which you observed this?
>>
>>
>> - missing output formats: parquet support to write generic Rows not
>> very well supported (at least out of the box) [1]
>>
>> Would you be up to opening up JIRAs for what you think is missing (if
>> there isn’t one already)?
>>
>>
>> - progress monitoring: for example in the ES connector there's no way
>> (apart from using accumulators) to monitor the progress of the indexing
>>
>> Maybe we can add some built-in metric in the ES sink connector that
>> tracks the number of successfully indexed elements, which can then be
>> queried from the REST API / Web UI. That wouldn’t be too much effort. What
>> do you think, would that be useful for your case?
>> Would be happy to hear your thoughts on this!
>>
>> Cheers,
>> Gordon
>>
>>
>> On 12 September 2017 at 11:36:27 AM, Flavio Pompermaier (
>> pompermaier@okkam.it) wrote:
>>
>> For the moment I give up with streaming...too many missing/unclear
>> features wrt batch.
>> For example:
>>
>> - checkpointing: it's not clear which checkpointing system to use and
>> how to tune/monitor it and avoid OOM exceptions. Moreover is it really
>> necessary to use it? For example if I read a file from HDFS and I don't
>> have a checkpoint it could be ok to re-run the job on all the data in case
>> of errors (i.e. the stream is managed like a batch)
>> - cleanup: BucketingSink doesn't always move to final state
>> - missing output formats: parquet support to write generic Rows not
>> very well supported (at least out of the box) [1]
>> - progress monitoring: for example in the ES connector there's no way
>> (apart from using accumulators) to monitor the progress of the indexing
>>
>> [1] https://stackoverflow.com/questions/41144659/flink-avro-
>> parquet-writer-in-rollingsink
>>
>> Maybe I'm wrong with those points but the attempt to replace my current
>> batch system with a streaming one had no luck with those points.
>>
>> Best,
>> Flavio
>>
>> On Fri, Sep 8, 2017 at 5:29 PM, Aljoscha Krettek <al...@apache.org>
>> wrote:
>>
>>> Hi,
>>>
>>> Expanding a bit on Kostas' answer. Yes, your analysis is correct, the
>>> problem is that the job is shutting down before a last checkpoint can
>>> "confirm" the written bucket data by moving it to the final state. The
>>> problem, as Kostas noted is that a user function (and thus also
>>> BucketingSink) does not know whether close() is being called because of a
>>> failure or because normal job shutdown. Therefore, we cannot move data to
>>> the final stage there.
>>>
>>> Once we have the issue that Kostas posted resolve we can also resolve
>>> this problem for the BucketingSink.
>>>
>>> Best,
>>> Aljoscha
>>>
>>> On 8. Sep 2017, at 16:48, Kostas Kloudas <k....@data-artisans.com>
>>> wrote:
>>>
>>> Hi Flavio,
>>>
>>> If I understand correctly, I think you bumped into this issue:
>>> https://issues.apache.org/jira/browse/FLINK-2646
>>>
>>> There is also a similar discussion on the BucketingSink here:
>>> http://apache-flink-mailing-list-archive.1008284.n3.nabble.c
>>> om/DISCUSS-Adding-a-dispose-method-in-the-RichFunction-td144
>>> 66.html#a14468
>>>
>>> Kostas
>>>
>>> On Sep 8, 2017, at 4:27 PM, Flavio Pompermaier <po...@okkam.it>
>>> wrote:
>>>
>>> Hi to all,
>>> I'm trying to test a streaming job but the files written by
>>> the BucketingSink are never finalized (remains into the pending state).
>>> Is this caused by the fact that the job finishes before the checkpoint?
>>> Shouldn't the sink properly close anyway?
>>>
>>> This is my code:
>>>
>>> @Test
>>> public void testBucketingSink() throws Exception {
>>> final StreamExecutionEnvironment senv =
>>> StreamExecutionEnvironment.getExecutionEnvironment();
>>> final StreamTableEnvironment tEnv = TableEnvironment.getTableEnvir
>>> onment(senv);
>>> senv.enableCheckpointing(5000);
>>> DataStream<String> testStream = senv.fromElements(//
>>> "1,aaa,white", //
>>> "2,bbb,gray", //
>>> "3,ccc,white", //
>>> "4,bbb,gray", //
>>> "5,bbb,gray" //
>>> );
>>> final RowTypeInfo rtf = new RowTypeInfo(
>>> BasicTypeInfo.STRING_TYPE_INFO,
>>> BasicTypeInfo.STRING_TYPE_INFO,
>>> BasicTypeInfo.STRING_TYPE_INFO);
>>> DataStream<Row> rows = testStream.map(new MapFunction<String, Row>()
>>> {
>>>
>>> private static final long serialVersionUID = 1L;
>>>
>>> @Override
>>> public Row map(String str) throws Exception {
>>> String[] split = str.split(Pattern.quote(","));
>>> Row ret = new Row(3);
>>> ret.setField(0, split[0]);
>>> ret.setField(1, split[1]);
>>> ret.setField(2, split[2]);
>>> return ret;
>>> }
>>> }).returns(rtf);
>>>
>>> String columnNames = "id,value,state";
>>> final String dsName = "test";
>>> tEnv.registerDataStream(dsName, rows, columnNames);
>>> final String whiteAreaFilter = "state = 'white'";
>>> DataStream<Row> grayArea = rows;
>>> DataStream<Row> whiteArea = null;
>>> if (whiteAreaFilter != null) {
>>> String sql = "SELECT *, (%s) as _WHITE FROM %s";
>>> sql = String.format(sql, whiteAreaFilter, dsName);
>>> Table table = tEnv.sql(sql);
>>> grayArea = tEnv.toDataStream(table.where(
>>> "!_WHITE").select(columnNames), rtf);
>>> DataStream<Row> nw = tEnv.toDataStream(table.where("_WHITE").select(columnNames),
>>> rtf);
>>> whiteArea = whiteArea == null ? nw : whiteArea.union(nw);
>>> }
>>> Writer<Row> bucketSinkwriter = new RowCsvWriter("UTF-8", "\t", "\n");
>>>
>>> String datasetWhiteDir = "/tmp/bucket/white";
>>> BucketingSink<Row> whiteAreaSink = new BucketingSink<>(datasetWhiteDi
>>> r.toString());
>>> whiteAreaSink.setWriter(bucketSinkwriter);
>>> whiteAreaSink.setBatchSize(10);
>>> whiteArea.addSink(whiteAreaSink);
>>>
>>> String datasetGrayDir = "/tmp/bucket/gray";
>>> BucketingSink<Row> grayAreaSink = new BucketingSink<>(datasetGrayDir
>>> .toString());
>>> grayAreaSink.setWriter(bucketSinkwriter);
>>> grayAreaSink.setBatchSize(10);
>>> grayArea.addSink(grayAreaSink);
>>>
>>> JobExecutionResult jobInfo = senv.execute("Buketing sink test ");
>>> System.out.printf("Job took %s minutes",
>>> jobInfo.getNetRuntime(TimeUnit.MINUTES));
>>> }
>>>
>>>
>>>
>>>
>>>
>>>
>>>
>>> public class RowCsvWriter extends StreamWriterBase<Row> {
>>> private static final long serialVersionUID = 1L;
>>>
>>> private final String charsetName;
>>> private transient Charset charset;
>>> private String fieldDelimiter;
>>> private String recordDelimiter;
>>> private boolean allowNullValues = true;
>>> private boolean quoteStrings = false;
>>>
>>> /**
>>> * Creates a new {@code StringWriter} that uses {@code "UTF-8"}
>>> charset to convert strings to
>>> * bytes.
>>> */
>>> public RowCsvWriter() {
>>> this("UTF-8", CsvOutputFormat.DEFAULT_FIELD_DELIMITER,
>>> CsvOutputFormat.DEFAULT_LINE_DELIMITER);
>>> }
>>>
>>> /**
>>> * Creates a new {@code StringWriter} that uses the given charset to
>>> convert strings to bytes.
>>> *
>>> * @param charsetName Name of the charset to be used, must be valid
>>> input for
>>> * {@code Charset.forName(charsetName)}
>>> */
>>> public RowCsvWriter(String charsetName, String fieldDelimiter, String
>>> recordDelimiter) {
>>> this.charsetName = charsetName;
>>> this.fieldDelimiter = fieldDelimiter;
>>> this.recordDelimiter = recordDelimiter;
>>> }
>>>
>>> @Override
>>> public void open(FileSystem fs, Path path) throws IOException {
>>> super.open(fs, path);
>>>
>>> try {
>>> this.charset = Charset.forName(charsetName);
>>> } catch (IllegalCharsetNameException ex) {
>>> throw new IOException("The charset " + charsetName + " is not
>>> valid.", ex);
>>> } catch (UnsupportedCharsetException ex) {
>>> throw new IOException("The charset " + charsetName + " is not
>>> supported.", ex);
>>> }
>>> }
>>>
>>> @Override
>>> public void write(Row element) throws IOException {
>>> FSDataOutputStream outputStream = getStream();
>>> writeRow(element, outputStream);
>>> }
>>>
>>> private void writeRow(Row element, FSDataOutputStream out) throws
>>> IOException {
>>> int numFields = element.getArity();
>>>
>>> for (int i = 0; i < numFields; i++) {
>>> Object obj = element.getField(i);
>>> if (obj != null) {
>>> if (i != 0) {
>>> out.write(this.fieldDelimiter.getBytes(charset));
>>> }
>>>
>>> if (quoteStrings) {
>>> if (obj instanceof String || obj instanceof StringValue) {
>>> out.write('"');
>>> out.write(obj.toString().getBytes(charset));
>>> out.write('"');
>>> } else {
>>> out.write(obj.toString().getBytes(charset));
>>> }
>>> } else {
>>> out.write(obj.toString().getBytes(charset));
>>> }
>>> } else {
>>> if (this.allowNullValues) {
>>> if (i != 0) {
>>> out.write(this.fieldDelimiter.getBytes(charset));
>>> }
>>> } else {
>>> throw new RuntimeException("Cannot write tuple with <null>
>>> value at position: " + i);
>>> }
>>> }
>>> }
>>>
>>> // add the record delimiter
>>> out.write(this.recordDelimiter.getBytes(charset));
>>> }
>>>
>>> @Override
>>> public Writer<Row> duplicate() {
>>> return new RowCsvWriter(charsetName, fieldDelimiter,
>>> recordDelimiter);
>>> }
>>> }
>>>
>>>
>>>
>>> Any help is appreciated,
>>> Flavio
>>>
>>>
>>>
>>>
>>
>
Re: BucketingSink never closed
Posted by Flavio Pompermaier <po...@okkam.it>.
Hi Gordon,
thanks for your feedback. The main problem for me is that moving from batch
to stream should be much easier IMHO.
Rows should be a first class citizen in Flink and should be VERY easy to
read/write them, while at the moment it seems that Tuples are the
dominating type...I don't want to write a serializer/outputFormat to
persist Rows as Parquet, Avro, Thrift, OCR, Kudu, Hive, etc..I expect to
have some already existing (and mantained) connector already available
somewhere. The case of the Parquet Rollink sink is just an example.
Regarding state backends I think that its not so easy to understand how to
design and monitor it properly: there are many parameters/variables to take
into account and it would be helpful to have a proper hands-on training
course/certification about this...
About ES indexing monitoring see my discussion with Chesnay at
http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Streaming-job-monitoring-td13583.html:
what I need is just to have recordsIn/recordsOut reflecting real values.
Best,
Flavio
On Wed, Sep 13, 2017 at 10:56 AM, Tzu-Li (Gordon) Tai <tz...@apache.org>
wrote:
> Hi Flavio,
>
> Let me try to understand / look at some of the problems you have
> encountered.
>
>
> - checkpointing: it's not clear which checkpointing system to use and
> how to tune/monitor it and avoid OOM exceptions.
>
> What do you mean be which "checkpointing system” to use? Do you mean state
> backends? Typically, you would only get OOM exceptions for memory-backed
> state backends if the state size exceeds the memory capacity. State sizes
> can be queried from the REST APIs / Web UI.
>
>
> - cleanup: BucketingSink doesn't always move to final state
>
> This sounds like a bug that we should look into. Do you have any logs on
> which you observed this?
>
>
> - missing output formats: parquet support to write generic Rows not
> very well supported (at least out of the box) [1]
>
> Would you be up to opening up JIRAs for what you think is missing (if
> there isn’t one already)?
>
>
> - progress monitoring: for example in the ES connector there's no way
> (apart from using accumulators) to monitor the progress of the indexing
>
> Maybe we can add some built-in metric in the ES sink connector that tracks
> the number of successfully indexed elements, which can then be queried from
> the REST API / Web UI. That wouldn’t be too much effort. What do you think,
> would that be useful for your case?
> Would be happy to hear your thoughts on this!
>
> Cheers,
> Gordon
>
>
> On 12 September 2017 at 11:36:27 AM, Flavio Pompermaier (
> pompermaier@okkam.it) wrote:
>
> For the moment I give up with streaming...too many missing/unclear
> features wrt batch.
> For example:
>
> - checkpointing: it's not clear which checkpointing system to use and
> how to tune/monitor it and avoid OOM exceptions. Moreover is it really
> necessary to use it? For example if I read a file from HDFS and I don't
> have a checkpoint it could be ok to re-run the job on all the data in case
> of errors (i.e. the stream is managed like a batch)
> - cleanup: BucketingSink doesn't always move to final state
> - missing output formats: parquet support to write generic Rows not
> very well supported (at least out of the box) [1]
> - progress monitoring: for example in the ES connector there's no way
> (apart from using accumulators) to monitor the progress of the indexing
>
> [1] https://stackoverflow.com/questions/41144659/flink-avro-
> parquet-writer-in-rollingsink
>
> Maybe I'm wrong with those points but the attempt to replace my current
> batch system with a streaming one had no luck with those points.
>
> Best,
> Flavio
>
> On Fri, Sep 8, 2017 at 5:29 PM, Aljoscha Krettek <al...@apache.org>
> wrote:
>
>> Hi,
>>
>> Expanding a bit on Kostas' answer. Yes, your analysis is correct, the
>> problem is that the job is shutting down before a last checkpoint can
>> "confirm" the written bucket data by moving it to the final state. The
>> problem, as Kostas noted is that a user function (and thus also
>> BucketingSink) does not know whether close() is being called because of a
>> failure or because normal job shutdown. Therefore, we cannot move data to
>> the final stage there.
>>
>> Once we have the issue that Kostas posted resolve we can also resolve
>> this problem for the BucketingSink.
>>
>> Best,
>> Aljoscha
>>
>> On 8. Sep 2017, at 16:48, Kostas Kloudas <k....@data-artisans.com>
>> wrote:
>>
>> Hi Flavio,
>>
>> If I understand correctly, I think you bumped into this issue:
>> https://issues.apache.org/jira/browse/FLINK-2646
>>
>> There is also a similar discussion on the BucketingSink here:
>> http://apache-flink-mailing-list-archive.1008284.n3.nabble.
>> com/DISCUSS-Adding-a-dispose-method-in-the-RichFunction-
>> td14466.html#a14468
>>
>> Kostas
>>
>> On Sep 8, 2017, at 4:27 PM, Flavio Pompermaier <po...@okkam.it>
>> wrote:
>>
>> Hi to all,
>> I'm trying to test a streaming job but the files written by
>> the BucketingSink are never finalized (remains into the pending state).
>> Is this caused by the fact that the job finishes before the checkpoint?
>> Shouldn't the sink properly close anyway?
>>
>> This is my code:
>>
>> @Test
>> public void testBucketingSink() throws Exception {
>> final StreamExecutionEnvironment senv = StreamExecutionEnvironment.get
>> ExecutionEnvironment();
>> final StreamTableEnvironment tEnv = TableEnvironment.getTableEnvir
>> onment(senv);
>> senv.enableCheckpointing(5000);
>> DataStream<String> testStream = senv.fromElements(//
>> "1,aaa,white", //
>> "2,bbb,gray", //
>> "3,ccc,white", //
>> "4,bbb,gray", //
>> "5,bbb,gray" //
>> );
>> final RowTypeInfo rtf = new RowTypeInfo(
>> BasicTypeInfo.STRING_TYPE_INFO,
>> BasicTypeInfo.STRING_TYPE_INFO,
>> BasicTypeInfo.STRING_TYPE_INFO);
>> DataStream<Row> rows = testStream.map(new MapFunction<String, Row>() {
>>
>> private static final long serialVersionUID = 1L;
>>
>> @Override
>> public Row map(String str) throws Exception {
>> String[] split = str.split(Pattern.quote(","));
>> Row ret = new Row(3);
>> ret.setField(0, split[0]);
>> ret.setField(1, split[1]);
>> ret.setField(2, split[2]);
>> return ret;
>> }
>> }).returns(rtf);
>>
>> String columnNames = "id,value,state";
>> final String dsName = "test";
>> tEnv.registerDataStream(dsName, rows, columnNames);
>> final String whiteAreaFilter = "state = 'white'";
>> DataStream<Row> grayArea = rows;
>> DataStream<Row> whiteArea = null;
>> if (whiteAreaFilter != null) {
>> String sql = "SELECT *, (%s) as _WHITE FROM %s";
>> sql = String.format(sql, whiteAreaFilter, dsName);
>> Table table = tEnv.sql(sql);
>> grayArea = tEnv.toDataStream(table.where(
>> "!_WHITE").select(columnNames), rtf);
>> DataStream<Row> nw = tEnv.toDataStream(table.where("_WHITE").select(columnNames),
>> rtf);
>> whiteArea = whiteArea == null ? nw : whiteArea.union(nw);
>> }
>> Writer<Row> bucketSinkwriter = new RowCsvWriter("UTF-8", "\t", "\n");
>>
>> String datasetWhiteDir = "/tmp/bucket/white";
>> BucketingSink<Row> whiteAreaSink = new BucketingSink<>(datasetWhiteDi
>> r.toString());
>> whiteAreaSink.setWriter(bucketSinkwriter);
>> whiteAreaSink.setBatchSize(10);
>> whiteArea.addSink(whiteAreaSink);
>>
>> String datasetGrayDir = "/tmp/bucket/gray";
>> BucketingSink<Row> grayAreaSink = new BucketingSink<>(datasetGrayDir
>> .toString());
>> grayAreaSink.setWriter(bucketSinkwriter);
>> grayAreaSink.setBatchSize(10);
>> grayArea.addSink(grayAreaSink);
>>
>> JobExecutionResult jobInfo = senv.execute("Buketing sink test ");
>> System.out.printf("Job took %s minutes",
>> jobInfo.getNetRuntime(TimeUnit.MINUTES));
>> }
>>
>>
>>
>>
>>
>>
>>
>> public class RowCsvWriter extends StreamWriterBase<Row> {
>> private static final long serialVersionUID = 1L;
>>
>> private final String charsetName;
>> private transient Charset charset;
>> private String fieldDelimiter;
>> private String recordDelimiter;
>> private boolean allowNullValues = true;
>> private boolean quoteStrings = false;
>>
>> /**
>> * Creates a new {@code StringWriter} that uses {@code "UTF-8"} charset
>> to convert strings to
>> * bytes.
>> */
>> public RowCsvWriter() {
>> this("UTF-8", CsvOutputFormat.DEFAULT_FIELD_DELIMITER,
>> CsvOutputFormat.DEFAULT_LINE_DELIMITER);
>> }
>>
>> /**
>> * Creates a new {@code StringWriter} that uses the given charset to
>> convert strings to bytes.
>> *
>> * @param charsetName Name of the charset to be used, must be valid
>> input for
>> * {@code Charset.forName(charsetName)}
>> */
>> public RowCsvWriter(String charsetName, String fieldDelimiter, String
>> recordDelimiter) {
>> this.charsetName = charsetName;
>> this.fieldDelimiter = fieldDelimiter;
>> this.recordDelimiter = recordDelimiter;
>> }
>>
>> @Override
>> public void open(FileSystem fs, Path path) throws IOException {
>> super.open(fs, path);
>>
>> try {
>> this.charset = Charset.forName(charsetName);
>> } catch (IllegalCharsetNameException ex) {
>> throw new IOException("The charset " + charsetName + " is not
>> valid.", ex);
>> } catch (UnsupportedCharsetException ex) {
>> throw new IOException("The charset " + charsetName + " is not
>> supported.", ex);
>> }
>> }
>>
>> @Override
>> public void write(Row element) throws IOException {
>> FSDataOutputStream outputStream = getStream();
>> writeRow(element, outputStream);
>> }
>>
>> private void writeRow(Row element, FSDataOutputStream out) throws
>> IOException {
>> int numFields = element.getArity();
>>
>> for (int i = 0; i < numFields; i++) {
>> Object obj = element.getField(i);
>> if (obj != null) {
>> if (i != 0) {
>> out.write(this.fieldDelimiter.getBytes(charset));
>> }
>>
>> if (quoteStrings) {
>> if (obj instanceof String || obj instanceof StringValue) {
>> out.write('"');
>> out.write(obj.toString().getBytes(charset));
>> out.write('"');
>> } else {
>> out.write(obj.toString().getBytes(charset));
>> }
>> } else {
>> out.write(obj.toString().getBytes(charset));
>> }
>> } else {
>> if (this.allowNullValues) {
>> if (i != 0) {
>> out.write(this.fieldDelimiter.getBytes(charset));
>> }
>> } else {
>> throw new RuntimeException("Cannot write tuple with <null>
>> value at position: " + i);
>> }
>> }
>> }
>>
>> // add the record delimiter
>> out.write(this.recordDelimiter.getBytes(charset));
>> }
>>
>> @Override
>> public Writer<Row> duplicate() {
>> return new RowCsvWriter(charsetName, fieldDelimiter, recordDelimiter);
>> }
>> }
>>
>>
>>
>> Any help is appreciated,
>> Flavio
>>
>>
>>
>>
>
Re: BucketingSink never closed
Posted by "Tzu-Li (Gordon) Tai" <tz...@apache.org>.
Hi Flavio,
Let me try to understand / look at some of the problems you have encountered.
checkpointing: it's not clear which checkpointing system to use and how to tune/monitor it and avoid OOM exceptions.
What do you mean be which "checkpointing system” to use? Do you mean state backends? Typically, you would only get OOM exceptions for memory-backed state backends if the state size exceeds the memory capacity. State sizes can be queried from the REST APIs / Web UI.
cleanup: BucketingSink doesn't always move to final state
This sounds like a bug that we should look into. Do you have any logs on which you observed this?
missing output formats: parquet support to write generic Rows not very well supported (at least out of the box) [1]
Would you be up to opening up JIRAs for what you think is missing (if there isn’t one already)?
progress monitoring: for example in the ES connector there's no way (apart from using accumulators) to monitor the progress of the indexing
Maybe we can add some built-in metric in the ES sink connector that tracks the number of successfully indexed elements, which can then be queried from the REST API / Web UI. That wouldn’t be too much effort. What do you think, would that be useful for your case?
Would be happy to hear your thoughts on this!
Cheers,
Gordon
On 12 September 2017 at 11:36:27 AM, Flavio Pompermaier (pompermaier@okkam.it) wrote:
For the moment I give up with streaming...too many missing/unclear features wrt batch.
For example:
checkpointing: it's not clear which checkpointing system to use and how to tune/monitor it and avoid OOM exceptions. Moreover is it really necessary to use it? For example if I read a file from HDFS and I don't have a checkpoint it could be ok to re-run the job on all the data in case of errors (i.e. the stream is managed like a batch)
cleanup: BucketingSink doesn't always move to final state
missing output formats: parquet support to write generic Rows not very well supported (at least out of the box) [1]
progress monitoring: for example in the ES connector there's no way (apart from using accumulators) to monitor the progress of the indexing
[1] https://stackoverflow.com/questions/41144659/flink-avro-parquet-writer-in-rollingsink
Maybe I'm wrong with those points but the attempt to replace my current batch system with a streaming one had no luck with those points.
Best,
Flavio
On Fri, Sep 8, 2017 at 5:29 PM, Aljoscha Krettek <al...@apache.org> wrote:
Hi,
Expanding a bit on Kostas' answer. Yes, your analysis is correct, the problem is that the job is shutting down before a last checkpoint can "confirm" the written bucket data by moving it to the final state. The problem, as Kostas noted is that a user function (and thus also BucketingSink) does not know whether close() is being called because of a failure or because normal job shutdown. Therefore, we cannot move data to the final stage there.
Once we have the issue that Kostas posted resolve we can also resolve this problem for the BucketingSink.
Best,
Aljoscha
On 8. Sep 2017, at 16:48, Kostas Kloudas <k....@data-artisans.com> wrote:
Hi Flavio,
If I understand correctly, I think you bumped into this issue: https://issues.apache.org/jira/browse/FLINK-2646
There is also a similar discussion on the BucketingSink here:
http://apache-flink-mailing-list-archive.1008284.n3.nabble.com/DISCUSS-Adding-a-dispose-method-in-the-RichFunction-td14466.html#a14468
Kostas
On Sep 8, 2017, at 4:27 PM, Flavio Pompermaier <po...@okkam.it> wrote:
Hi to all,
I'm trying to test a streaming job but the files written by the BucketingSink are never finalized (remains into the pending state).
Is this caused by the fact that the job finishes before the checkpoint?
Shouldn't the sink properly close anyway?
This is my code:
@Test
public void testBucketingSink() throws Exception {
final StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
final StreamTableEnvironment tEnv = TableEnvironment.getTableEnvironment(senv);
senv.enableCheckpointing(5000);
DataStream<String> testStream = senv.fromElements(//
"1,aaa,white", //
"2,bbb,gray", //
"3,ccc,white", //
"4,bbb,gray", //
"5,bbb,gray" //
);
final RowTypeInfo rtf = new RowTypeInfo(
BasicTypeInfo.STRING_TYPE_INFO,
BasicTypeInfo.STRING_TYPE_INFO,
BasicTypeInfo.STRING_TYPE_INFO);
DataStream<Row> rows = testStream.map(new MapFunction<String, Row>() {
private static final long serialVersionUID = 1L;
@Override
public Row map(String str) throws Exception {
String[] split = str.split(Pattern.quote(","));
Row ret = new Row(3);
ret.setField(0, split[0]);
ret.setField(1, split[1]);
ret.setField(2, split[2]);
return ret;
}
}).returns(rtf);
String columnNames = "id,value,state";
final String dsName = "test";
tEnv.registerDataStream(dsName, rows, columnNames);
final String whiteAreaFilter = "state = 'white'";
DataStream<Row> grayArea = rows;
DataStream<Row> whiteArea = null;
if (whiteAreaFilter != null) {
String sql = "SELECT *, (%s) as _WHITE FROM %s";
sql = String.format(sql, whiteAreaFilter, dsName);
Table table = tEnv.sql(sql);
grayArea = tEnv.toDataStream(table.where("!_WHITE").select(columnNames), rtf);
DataStream<Row> nw = tEnv.toDataStream(table.where("_WHITE").select(columnNames), rtf);
whiteArea = whiteArea == null ? nw : whiteArea.union(nw);
}
Writer<Row> bucketSinkwriter = new RowCsvWriter("UTF-8", "\t", "\n");
String datasetWhiteDir = "/tmp/bucket/white";
BucketingSink<Row> whiteAreaSink = new BucketingSink<>(datasetWhiteDir.toString());
whiteAreaSink.setWriter(bucketSinkwriter);
whiteAreaSink.setBatchSize(10);
whiteArea.addSink(whiteAreaSink);
String datasetGrayDir = "/tmp/bucket/gray";
BucketingSink<Row> grayAreaSink = new BucketingSink<>(datasetGrayDir.toString());
grayAreaSink.setWriter(bucketSinkwriter);
grayAreaSink.setBatchSize(10);
grayArea.addSink(grayAreaSink);
JobExecutionResult jobInfo = senv.execute("Buketing sink test ");
System.out.printf("Job took %s minutes", jobInfo.getNetRuntime(TimeUnit.MINUTES));
}
public class RowCsvWriter extends StreamWriterBase<Row> {
private static final long serialVersionUID = 1L;
private final String charsetName;
private transient Charset charset;
private String fieldDelimiter;
private String recordDelimiter;
private boolean allowNullValues = true;
private boolean quoteStrings = false;
/**
* Creates a new {@code StringWriter} that uses {@code "UTF-8"} charset to convert strings to
* bytes.
*/
public RowCsvWriter() {
this("UTF-8", CsvOutputFormat.DEFAULT_FIELD_DELIMITER, CsvOutputFormat.DEFAULT_LINE_DELIMITER);
}
/**
* Creates a new {@code StringWriter} that uses the given charset to convert strings to bytes.
*
* @param charsetName Name of the charset to be used, must be valid input for
* {@code Charset.forName(charsetName)}
*/
public RowCsvWriter(String charsetName, String fieldDelimiter, String recordDelimiter) {
this.charsetName = charsetName;
this.fieldDelimiter = fieldDelimiter;
this.recordDelimiter = recordDelimiter;
}
@Override
public void open(FileSystem fs, Path path) throws IOException {
super.open(fs, path);
try {
this.charset = Charset.forName(charsetName);
} catch (IllegalCharsetNameException ex) {
throw new IOException("The charset " + charsetName + " is not valid.", ex);
} catch (UnsupportedCharsetException ex) {
throw new IOException("The charset " + charsetName + " is not supported.", ex);
}
}
@Override
public void write(Row element) throws IOException {
FSDataOutputStream outputStream = getStream();
writeRow(element, outputStream);
}
private void writeRow(Row element, FSDataOutputStream out) throws IOException {
int numFields = element.getArity();
for (int i = 0; i < numFields; i++) {
Object obj = element.getField(i);
if (obj != null) {
if (i != 0) {
out.write(this.fieldDelimiter.getBytes(charset));
}
if (quoteStrings) {
if (obj instanceof String || obj instanceof StringValue) {
out.write('"');
out.write(obj.toString().getBytes(charset));
out.write('"');
} else {
out.write(obj.toString().getBytes(charset));
}
} else {
out.write(obj.toString().getBytes(charset));
}
} else {
if (this.allowNullValues) {
if (i != 0) {
out.write(this.fieldDelimiter.getBytes(charset));
}
} else {
throw new RuntimeException("Cannot write tuple with <null> value at position: " + i);
}
}
}
// add the record delimiter
out.write(this.recordDelimiter.getBytes(charset));
}
@Override
public Writer<Row> duplicate() {
return new RowCsvWriter(charsetName, fieldDelimiter, recordDelimiter);
}
}
Any help is appreciated,
Flavio
Re: BucketingSink never closed
Posted by Flavio Pompermaier <po...@okkam.it>.
For the moment I give up with streaming...too many missing/unclear features
wrt batch.
For example:
- checkpointing: it's not clear which checkpointing system to use and
how to tune/monitor it and avoid OOM exceptions. Moreover is it really
necessary to use it? For example if I read a file from HDFS and I don't
have a checkpoint it could be ok to re-run the job on all the data in case
of errors (i.e. the stream is managed like a batch)
- cleanup: BucketingSink doesn't always move to final state
- missing output formats: parquet support to write generic Rows not very
well supported (at least out of the box) [1]
- progress monitoring: for example in the ES connector there's no way
(apart from using accumulators) to monitor the progress of the indexing
[1]
https://stackoverflow.com/questions/41144659/flink-avro-parquet-writer-in-rollingsink
Maybe I'm wrong with those points but the attempt to replace my current
batch system with a streaming one had no luck with those points.
Best,
Flavio
On Fri, Sep 8, 2017 at 5:29 PM, Aljoscha Krettek <al...@apache.org>
wrote:
> Hi,
>
> Expanding a bit on Kostas' answer. Yes, your analysis is correct, the
> problem is that the job is shutting down before a last checkpoint can
> "confirm" the written bucket data by moving it to the final state. The
> problem, as Kostas noted is that a user function (and thus also
> BucketingSink) does not know whether close() is being called because of a
> failure or because normal job shutdown. Therefore, we cannot move data to
> the final stage there.
>
> Once we have the issue that Kostas posted resolve we can also resolve this
> problem for the BucketingSink.
>
> Best,
> Aljoscha
>
> On 8. Sep 2017, at 16:48, Kostas Kloudas <k....@data-artisans.com>
> wrote:
>
> Hi Flavio,
>
> If I understand correctly, I think you bumped into this issue:
> https://issues.apache.org/jira/browse/FLINK-2646
>
> There is also a similar discussion on the BucketingSink here:
> http://apache-flink-mailing-list-archive.1008284.n3.
> nabble.com/DISCUSS-Adding-a-dispose-method-in-the-
> RichFunction-td14466.html#a14468
>
> Kostas
>
> On Sep 8, 2017, at 4:27 PM, Flavio Pompermaier <po...@okkam.it>
> wrote:
>
> Hi to all,
> I'm trying to test a streaming job but the files written by
> the BucketingSink are never finalized (remains into the pending state).
> Is this caused by the fact that the job finishes before the checkpoint?
> Shouldn't the sink properly close anyway?
>
> This is my code:
>
> @Test
> public void testBucketingSink() throws Exception {
> final StreamExecutionEnvironment senv = StreamExecutionEnvironment.
> getExecutionEnvironment();
> final StreamTableEnvironment tEnv = TableEnvironment.
> getTableEnvironment(senv);
> senv.enableCheckpointing(5000);
> DataStream<String> testStream = senv.fromElements(//
> "1,aaa,white", //
> "2,bbb,gray", //
> "3,ccc,white", //
> "4,bbb,gray", //
> "5,bbb,gray" //
> );
> final RowTypeInfo rtf = new RowTypeInfo(
> BasicTypeInfo.STRING_TYPE_INFO,
> BasicTypeInfo.STRING_TYPE_INFO,
> BasicTypeInfo.STRING_TYPE_INFO);
> DataStream<Row> rows = testStream.map(new MapFunction<String, Row>() {
>
> private static final long serialVersionUID = 1L;
>
> @Override
> public Row map(String str) throws Exception {
> String[] split = str.split(Pattern.quote(","));
> Row ret = new Row(3);
> ret.setField(0, split[0]);
> ret.setField(1, split[1]);
> ret.setField(2, split[2]);
> return ret;
> }
> }).returns(rtf);
>
> String columnNames = "id,value,state";
> final String dsName = "test";
> tEnv.registerDataStream(dsName, rows, columnNames);
> final String whiteAreaFilter = "state = 'white'";
> DataStream<Row> grayArea = rows;
> DataStream<Row> whiteArea = null;
> if (whiteAreaFilter != null) {
> String sql = "SELECT *, (%s) as _WHITE FROM %s";
> sql = String.format(sql, whiteAreaFilter, dsName);
> Table table = tEnv.sql(sql);
> grayArea = tEnv.toDataStream(table.where(
> "!_WHITE").select(columnNames), rtf);
> DataStream<Row> nw = tEnv.toDataStream(table.where("_WHITE").select(columnNames),
> rtf);
> whiteArea = whiteArea == null ? nw : whiteArea.union(nw);
> }
> Writer<Row> bucketSinkwriter = new RowCsvWriter("UTF-8", "\t", "\n");
>
> String datasetWhiteDir = "/tmp/bucket/white";
> BucketingSink<Row> whiteAreaSink = new BucketingSink<>(
> datasetWhiteDir.toString());
> whiteAreaSink.setWriter(bucketSinkwriter);
> whiteAreaSink.setBatchSize(10);
> whiteArea.addSink(whiteAreaSink);
>
> String datasetGrayDir = "/tmp/bucket/gray";
> BucketingSink<Row> grayAreaSink = new BucketingSink<>(
> datasetGrayDir.toString());
> grayAreaSink.setWriter(bucketSinkwriter);
> grayAreaSink.setBatchSize(10);
> grayArea.addSink(grayAreaSink);
>
> JobExecutionResult jobInfo = senv.execute("Buketing sink test ");
> System.out.printf("Job took %s minutes", jobInfo.getNetRuntime(
> TimeUnit.MINUTES));
> }
>
>
>
>
>
>
>
> public class RowCsvWriter extends StreamWriterBase<Row> {
> private static final long serialVersionUID = 1L;
>
> private final String charsetName;
> private transient Charset charset;
> private String fieldDelimiter;
> private String recordDelimiter;
> private boolean allowNullValues = true;
> private boolean quoteStrings = false;
>
> /**
> * Creates a new {@code StringWriter} that uses {@code "UTF-8"} charset
> to convert strings to
> * bytes.
> */
> public RowCsvWriter() {
> this("UTF-8", CsvOutputFormat.DEFAULT_FIELD_DELIMITER,
> CsvOutputFormat.DEFAULT_LINE_DELIMITER);
> }
>
> /**
> * Creates a new {@code StringWriter} that uses the given charset to
> convert strings to bytes.
> *
> * @param charsetName Name of the charset to be used, must be valid
> input for
> * {@code Charset.forName(charsetName)}
> */
> public RowCsvWriter(String charsetName, String fieldDelimiter, String
> recordDelimiter) {
> this.charsetName = charsetName;
> this.fieldDelimiter = fieldDelimiter;
> this.recordDelimiter = recordDelimiter;
> }
>
> @Override
> public void open(FileSystem fs, Path path) throws IOException {
> super.open(fs, path);
>
> try {
> this.charset = Charset.forName(charsetName);
> } catch (IllegalCharsetNameException ex) {
> throw new IOException("The charset " + charsetName + " is not
> valid.", ex);
> } catch (UnsupportedCharsetException ex) {
> throw new IOException("The charset " + charsetName + " is not
> supported.", ex);
> }
> }
>
> @Override
> public void write(Row element) throws IOException {
> FSDataOutputStream outputStream = getStream();
> writeRow(element, outputStream);
> }
>
> private void writeRow(Row element, FSDataOutputStream out) throws
> IOException {
> int numFields = element.getArity();
>
> for (int i = 0; i < numFields; i++) {
> Object obj = element.getField(i);
> if (obj != null) {
> if (i != 0) {
> out.write(this.fieldDelimiter.getBytes(charset));
> }
>
> if (quoteStrings) {
> if (obj instanceof String || obj instanceof StringValue) {
> out.write('"');
> out.write(obj.toString().getBytes(charset));
> out.write('"');
> } else {
> out.write(obj.toString().getBytes(charset));
> }
> } else {
> out.write(obj.toString().getBytes(charset));
> }
> } else {
> if (this.allowNullValues) {
> if (i != 0) {
> out.write(this.fieldDelimiter.getBytes(charset));
> }
> } else {
> throw new RuntimeException("Cannot write tuple with <null> value
> at position: " + i);
> }
> }
> }
>
> // add the record delimiter
> out.write(this.recordDelimiter.getBytes(charset));
> }
>
> @Override
> public Writer<Row> duplicate() {
> return new RowCsvWriter(charsetName, fieldDelimiter, recordDelimiter);
> }
> }
>
>
>
> Any help is appreciated,
> Flavio
>
>
>
>
Re: BucketingSink never closed
Posted by Aljoscha Krettek <al...@apache.org>.
Hi,
Expanding a bit on Kostas' answer. Yes, your analysis is correct, the problem is that the job is shutting down before a last checkpoint can "confirm" the written bucket data by moving it to the final state. The problem, as Kostas noted is that a user function (and thus also BucketingSink) does not know whether close() is being called because of a failure or because normal job shutdown. Therefore, we cannot move data to the final stage there.
Once we have the issue that Kostas posted resolve we can also resolve this problem for the BucketingSink.
Best,
Aljoscha
> On 8. Sep 2017, at 16:48, Kostas Kloudas <k....@data-artisans.com> wrote:
>
> Hi Flavio,
>
> If I understand correctly, I think you bumped into this issue: https://issues.apache.org/jira/browse/FLINK-2646 <https://issues.apache.org/jira/browse/FLINK-2646>
>
> There is also a similar discussion on the BucketingSink here:
> http://apache-flink-mailing-list-archive.1008284.n3.nabble.com/DISCUSS-Adding-a-dispose-method-in-the-RichFunction-td14466.html#a14468 <http://apache-flink-mailing-list-archive.1008284.n3.nabble.com/DISCUSS-Adding-a-dispose-method-in-the-RichFunction-td14466.html#a14468>
>
> Kostas
>
>> On Sep 8, 2017, at 4:27 PM, Flavio Pompermaier <pompermaier@okkam.it <ma...@okkam.it>> wrote:
>>
>> Hi to all,
>> I'm trying to test a streaming job but the files written by the BucketingSink are never finalized (remains into the pending state).
>> Is this caused by the fact that the job finishes before the checkpoint?
>> Shouldn't the sink properly close anyway?
>>
>> This is my code:
>>
>> @Test
>> public void testBucketingSink() throws Exception {
>> final StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
>> final StreamTableEnvironment tEnv = TableEnvironment.getTableEnvironment(senv);
>> senv.enableCheckpointing(5000);
>> DataStream<String> testStream = senv.fromElements(//
>> "1,aaa,white", //
>> "2,bbb,gray", //
>> "3,ccc,white", //
>> "4,bbb,gray", //
>> "5,bbb,gray" //
>> );
>> final RowTypeInfo rtf = new RowTypeInfo(
>> BasicTypeInfo.STRING_TYPE_INFO,
>> BasicTypeInfo.STRING_TYPE_INFO,
>> BasicTypeInfo.STRING_TYPE_INFO);
>> DataStream<Row> rows = testStream.map(new MapFunction<String, Row>() {
>>
>> private static final long serialVersionUID = 1L;
>>
>> @Override
>> public Row map(String str) throws Exception {
>> String[] split = str.split(Pattern.quote(","));
>> Row ret = new Row(3);
>> ret.setField(0, split[0]);
>> ret.setField(1, split[1]);
>> ret.setField(2, split[2]);
>> return ret;
>> }
>> }).returns(rtf);
>>
>> String columnNames = "id,value,state";
>> final String dsName = "test";
>> tEnv.registerDataStream(dsName, rows, columnNames);
>> final String whiteAreaFilter = "state = 'white'";
>> DataStream<Row> grayArea = rows;
>> DataStream<Row> whiteArea = null;
>> if (whiteAreaFilter != null) {
>> String sql = "SELECT *, (%s) as _WHITE FROM %s";
>> sql = String.format(sql, whiteAreaFilter, dsName);
>> Table table = tEnv.sql(sql);
>> grayArea = tEnv.toDataStream(table.where("!_WHITE").select(columnNames), rtf);
>> DataStream<Row> nw = tEnv.toDataStream(table.where("_WHITE").select(columnNames), rtf);
>> whiteArea = whiteArea == null ? nw : whiteArea.union(nw);
>> }
>> Writer<Row> bucketSinkwriter = new RowCsvWriter("UTF-8", "\t", "\n");
>>
>> String datasetWhiteDir = "/tmp/bucket/white";
>> BucketingSink<Row> whiteAreaSink = new BucketingSink<>(datasetWhiteDir.toString());
>> whiteAreaSink.setWriter(bucketSinkwriter);
>> whiteAreaSink.setBatchSize(10);
>> whiteArea.addSink(whiteAreaSink);
>>
>> String datasetGrayDir = "/tmp/bucket/gray";
>> BucketingSink<Row> grayAreaSink = new BucketingSink<>(datasetGrayDir.toString());
>> grayAreaSink.setWriter(bucketSinkwriter);
>> grayAreaSink.setBatchSize(10);
>> grayArea.addSink(grayAreaSink);
>>
>> JobExecutionResult jobInfo = senv.execute("Buketing sink test ");
>> System.out.printf("Job took %s minutes", jobInfo.getNetRuntime(TimeUnit.MINUTES));
>> }
>>
>>
>>
>>
>>
>>
>>
>> public class RowCsvWriter extends StreamWriterBase<Row> {
>> private static final long serialVersionUID = 1L;
>>
>> private final String charsetName;
>> private transient Charset charset;
>> private String fieldDelimiter;
>> private String recordDelimiter;
>> private boolean allowNullValues = true;
>> private boolean quoteStrings = false;
>>
>> /**
>> * Creates a new {@code StringWriter} that uses {@code "UTF-8"} charset to convert strings to
>> * bytes.
>> */
>> public RowCsvWriter() {
>> this("UTF-8", CsvOutputFormat.DEFAULT_FIELD_DELIMITER, CsvOutputFormat.DEFAULT_LINE_DELIMITER);
>> }
>>
>> /**
>> * Creates a new {@code StringWriter} that uses the given charset to convert strings to bytes.
>> *
>> * @param charsetName Name of the charset to be used, must be valid input for
>> * {@code Charset.forName(charsetName)}
>> */
>> public RowCsvWriter(String charsetName, String fieldDelimiter, String recordDelimiter) {
>> this.charsetName = charsetName;
>> this.fieldDelimiter = fieldDelimiter;
>> this.recordDelimiter = recordDelimiter;
>> }
>>
>> @Override
>> public void open(FileSystem fs, Path path) throws IOException {
>> super.open(fs, path);
>>
>> try {
>> this.charset = Charset.forName(charsetName);
>> } catch (IllegalCharsetNameException ex) {
>> throw new IOException("The charset " + charsetName + " is not valid.", ex);
>> } catch (UnsupportedCharsetException ex) {
>> throw new IOException("The charset " + charsetName + " is not supported.", ex);
>> }
>> }
>>
>> @Override
>> public void write(Row element) throws IOException {
>> FSDataOutputStream outputStream = getStream();
>> writeRow(element, outputStream);
>> }
>>
>> private void writeRow(Row element, FSDataOutputStream out) throws IOException {
>> int numFields = element.getArity();
>>
>> for (int i = 0; i < numFields; i++) {
>> Object obj = element.getField(i);
>> if (obj != null) {
>> if (i != 0) {
>> out.write(this.fieldDelimiter.getBytes(charset));
>> }
>>
>> if (quoteStrings) {
>> if (obj instanceof String || obj instanceof StringValue) {
>> out.write('"');
>> out.write(obj.toString().getBytes(charset));
>> out.write('"');
>> } else {
>> out.write(obj.toString().getBytes(charset));
>> }
>> } else {
>> out.write(obj.toString().getBytes(charset));
>> }
>> } else {
>> if (this.allowNullValues) {
>> if (i != 0) {
>> out.write(this.fieldDelimiter.getBytes(charset));
>> }
>> } else {
>> throw new RuntimeException("Cannot write tuple with <null> value at position: " + i);
>> }
>> }
>> }
>>
>> // add the record delimiter
>> out.write(this.recordDelimiter.getBytes(charset));
>> }
>>
>> @Override
>> public Writer<Row> duplicate() {
>> return new RowCsvWriter(charsetName, fieldDelimiter, recordDelimiter);
>> }
>> }
>>
>>
>>
>> Any help is appreciated,
>> Flavio
>
Re: BucketingSink never closed
Posted by Kostas Kloudas <k....@data-artisans.com>.
Hi Flavio,
If I understand correctly, I think you bumped into this issue: https://issues.apache.org/jira/browse/FLINK-2646 <https://issues.apache.org/jira/browse/FLINK-2646>
There is also a similar discussion on the BucketingSink here:
http://apache-flink-mailing-list-archive.1008284.n3.nabble.com/DISCUSS-Adding-a-dispose-method-in-the-RichFunction-td14466.html#a14468 <http://apache-flink-mailing-list-archive.1008284.n3.nabble.com/DISCUSS-Adding-a-dispose-method-in-the-RichFunction-td14466.html#a14468>
Kostas
> On Sep 8, 2017, at 4:27 PM, Flavio Pompermaier <po...@okkam.it> wrote:
>
> Hi to all,
> I'm trying to test a streaming job but the files written by the BucketingSink are never finalized (remains into the pending state).
> Is this caused by the fact that the job finishes before the checkpoint?
> Shouldn't the sink properly close anyway?
>
> This is my code:
>
> @Test
> public void testBucketingSink() throws Exception {
> final StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
> final StreamTableEnvironment tEnv = TableEnvironment.getTableEnvironment(senv);
> senv.enableCheckpointing(5000);
> DataStream<String> testStream = senv.fromElements(//
> "1,aaa,white", //
> "2,bbb,gray", //
> "3,ccc,white", //
> "4,bbb,gray", //
> "5,bbb,gray" //
> );
> final RowTypeInfo rtf = new RowTypeInfo(
> BasicTypeInfo.STRING_TYPE_INFO,
> BasicTypeInfo.STRING_TYPE_INFO,
> BasicTypeInfo.STRING_TYPE_INFO);
> DataStream<Row> rows = testStream.map(new MapFunction<String, Row>() {
>
> private static final long serialVersionUID = 1L;
>
> @Override
> public Row map(String str) throws Exception {
> String[] split = str.split(Pattern.quote(","));
> Row ret = new Row(3);
> ret.setField(0, split[0]);
> ret.setField(1, split[1]);
> ret.setField(2, split[2]);
> return ret;
> }
> }).returns(rtf);
>
> String columnNames = "id,value,state";
> final String dsName = "test";
> tEnv.registerDataStream(dsName, rows, columnNames);
> final String whiteAreaFilter = "state = 'white'";
> DataStream<Row> grayArea = rows;
> DataStream<Row> whiteArea = null;
> if (whiteAreaFilter != null) {
> String sql = "SELECT *, (%s) as _WHITE FROM %s";
> sql = String.format(sql, whiteAreaFilter, dsName);
> Table table = tEnv.sql(sql);
> grayArea = tEnv.toDataStream(table.where("!_WHITE").select(columnNames), rtf);
> DataStream<Row> nw = tEnv.toDataStream(table.where("_WHITE").select(columnNames), rtf);
> whiteArea = whiteArea == null ? nw : whiteArea.union(nw);
> }
> Writer<Row> bucketSinkwriter = new RowCsvWriter("UTF-8", "\t", "\n");
>
> String datasetWhiteDir = "/tmp/bucket/white";
> BucketingSink<Row> whiteAreaSink = new BucketingSink<>(datasetWhiteDir.toString());
> whiteAreaSink.setWriter(bucketSinkwriter);
> whiteAreaSink.setBatchSize(10);
> whiteArea.addSink(whiteAreaSink);
>
> String datasetGrayDir = "/tmp/bucket/gray";
> BucketingSink<Row> grayAreaSink = new BucketingSink<>(datasetGrayDir.toString());
> grayAreaSink.setWriter(bucketSinkwriter);
> grayAreaSink.setBatchSize(10);
> grayArea.addSink(grayAreaSink);
>
> JobExecutionResult jobInfo = senv.execute("Buketing sink test ");
> System.out.printf("Job took %s minutes", jobInfo.getNetRuntime(TimeUnit.MINUTES));
> }
>
>
>
>
>
>
>
> public class RowCsvWriter extends StreamWriterBase<Row> {
> private static final long serialVersionUID = 1L;
>
> private final String charsetName;
> private transient Charset charset;
> private String fieldDelimiter;
> private String recordDelimiter;
> private boolean allowNullValues = true;
> private boolean quoteStrings = false;
>
> /**
> * Creates a new {@code StringWriter} that uses {@code "UTF-8"} charset to convert strings to
> * bytes.
> */
> public RowCsvWriter() {
> this("UTF-8", CsvOutputFormat.DEFAULT_FIELD_DELIMITER, CsvOutputFormat.DEFAULT_LINE_DELIMITER);
> }
>
> /**
> * Creates a new {@code StringWriter} that uses the given charset to convert strings to bytes.
> *
> * @param charsetName Name of the charset to be used, must be valid input for
> * {@code Charset.forName(charsetName)}
> */
> public RowCsvWriter(String charsetName, String fieldDelimiter, String recordDelimiter) {
> this.charsetName = charsetName;
> this.fieldDelimiter = fieldDelimiter;
> this.recordDelimiter = recordDelimiter;
> }
>
> @Override
> public void open(FileSystem fs, Path path) throws IOException {
> super.open(fs, path);
>
> try {
> this.charset = Charset.forName(charsetName);
> } catch (IllegalCharsetNameException ex) {
> throw new IOException("The charset " + charsetName + " is not valid.", ex);
> } catch (UnsupportedCharsetException ex) {
> throw new IOException("The charset " + charsetName + " is not supported.", ex);
> }
> }
>
> @Override
> public void write(Row element) throws IOException {
> FSDataOutputStream outputStream = getStream();
> writeRow(element, outputStream);
> }
>
> private void writeRow(Row element, FSDataOutputStream out) throws IOException {
> int numFields = element.getArity();
>
> for (int i = 0; i < numFields; i++) {
> Object obj = element.getField(i);
> if (obj != null) {
> if (i != 0) {
> out.write(this.fieldDelimiter.getBytes(charset));
> }
>
> if (quoteStrings) {
> if (obj instanceof String || obj instanceof StringValue) {
> out.write('"');
> out.write(obj.toString().getBytes(charset));
> out.write('"');
> } else {
> out.write(obj.toString().getBytes(charset));
> }
> } else {
> out.write(obj.toString().getBytes(charset));
> }
> } else {
> if (this.allowNullValues) {
> if (i != 0) {
> out.write(this.fieldDelimiter.getBytes(charset));
> }
> } else {
> throw new RuntimeException("Cannot write tuple with <null> value at position: " + i);
> }
> }
> }
>
> // add the record delimiter
> out.write(this.recordDelimiter.getBytes(charset));
> }
>
> @Override
> public Writer<Row> duplicate() {
> return new RowCsvWriter(charsetName, fieldDelimiter, recordDelimiter);
> }
> }
>
>
>
> Any help is appreciated,
> Flavio