Hi Rafi,
Have you enabled checkpointing for you job?
Cheers,
Kostas
On Thu, Mar 21, 2019 at 5:18 PM Rafi Aroch <rafi.ar...@gmail.com> wrote:
> Hi Piotr and Kostas,
>
> Thanks for your reply.
>
> The issue is that I don't see any committed files, only in-progress.
> I tried to debug the code for more details. I see that in *BulkPartWriter* I
> do reach the *write* methods and see events getting written, but I never
> reach the *closeForCommit*. I reach straight to the *close* function
> where all parts are disposed.
>
> In my job I have a finite stream (source is reading from parquet file/s).
> Doing some windowed aggregation and writing back to a parquet file.
> As far as I know, it should commit files during checkpoints and when the
> stream has finished. I did enabled checkpointing.
> I did verify that if I connect to other sinks, I see the events.
>
> Let me know if I can provide any further information that could be helpful.
>
> Would appreciate your help.
>
> Thanks,
> Rafi
>
>
> On Thu, Mar 21, 2019 at 5:20 PM Kostas Kloudas <kklou...@gmail.com> wrote:
>
>> Hi Rafi,
>>
>> Piotr is correct. In-progress files are not necessarily readable.
>> The valid files are the ones that are "committed" or finalized.
>>
>> Cheers,
>> Kostas
>>
>> On Thu, Mar 21, 2019 at 2:53 PM Piotr Nowojski <pi...@ververica.com>
>> wrote:
>>
>>> Hi,
>>>
>>> I’m not sure, but shouldn’t you be just reading committed files and
>>> ignore in-progress? Maybe Kostas could add more insight to this topic.
>>>
>>> Piotr Nowojski
>>>
>>> On 20 Mar 2019, at 12:23, Rafi Aroch <rafi.ar...@gmail.com> wrote:
>>>
>>> Hi,
>>>
>>> I'm trying to stream events in Prorobuf format into a parquet file.
>>> I looked into both streaming-file options: BucketingSink &
>>> StreamingFileSink.
>>> I first tried using the newer *StreamingFileSink* with the *forBulkFormat
>>> *API. I noticed there's currently support only for the Avro format with
>>> the *ParquetAvroWriters*.
>>> I followed the same convention as Avro and wrote a *ParquetProtoWriters*
>>> builder
>>> class:
>>>
>>> public class ParquetProtoWriters {
>>>
>>> private static final int pageSize = 64 * 1024;
>>>
>>> public static <T extends Message> ParquetWriterFactory<T> forType(final
>>> Class<T> protoClass) {
>>> final ParquetBuilder<T> builder = (out) ->
>>> createProtoParquetWriter(protoClass, out);
>>> return new ParquetWriterFactory<>(builder);
>>> }
>>>
>>> private static <T extends Message> ParquetWriter<T>
>>> createProtoParquetWriter(
>>> Class<T> type,
>>> OutputFile out) throws IOException {
>>>
>>> return ProtoParquetWriter.<T>builder(out)
>>> .withPageSize(pageSize)
>>> .withWriteMode(ParquetFileWriter.Mode.OVERWRITE)
>>> .withCompressionCodec(CompressionCodecName.SNAPPY)
>>> .withProtoClass(type)
>>> .build();
>>> }
>>> }
>>>
>>> And then I use it as follows:
>>>
>>> StreamingFileSink
>>> .forBulkFormat(new Path("some-path),
>>> ParquetProtoWriters.forType(SomeProtoType.class))
>>> .build();
>>>
>>> I ran tests on the *ParquetProtoWriters *itself and it writes
>>> everything properly and i'm able to read the files.
>>>
>>> When I use the sink as part of a job *I see illegal Parquet files
>>> created*:
>>>
>>> # parquet-tools cat
>>> .part-0-0.inprogress.b3a65d7f-f297-46bd-bdeb-ad24106ad3ea
>>> .part-0-0.inprogress.b3a65d7f-f297-46bd-bdeb-ad24106ad3ea is not a Parquet
>>> file (too small length: 4)
>>>
>>>
>>> Can anyone suggest what am I missing here?
>>>
>>> When trying to use the *BucketingSink*, I wrote a Writer class for
>>> Protobuf and everything worked perfectly:
>>>
>>> public class FlinkProtoParquetWriter<T extends MessageOrBuilder> implements
>>> Writer<T> {
>>>
>>> private static final long serialVersionUID = -975302556515811398L;
>>>
>>> private Path path;
>>> private Class<? extends Message> protoClass;
>>> private transient ParquetWriter<T> writer;
>>>
>>> private int position;
>>> private final CompressionCodecName compressionCodecName =
>>> CompressionCodecName.SNAPPY;
>>> private final int pageSize = 64 * 1024;
>>>
>>> public FlinkProtoParquetWriter(Class<? extends Message> protoClass) {
>>> this.protoClass = protoClass;
>>> }
>>>
>>> @Override
>>> public void open(FileSystem fs, Path path) throws IOException {
>>> this.position = 0;
>>> this.path = path;
>>>
>>> if (writer != null) {
>>> writer.close();
>>> }
>>>
>>> writer = createWriter();
>>> }
>>>
>>> @Override
>>> public long flush() throws IOException {
>>> Preconditions.checkNotNull(writer);
>>> position += writer.getDataSize();
>>> writer.close();
>>> writer = createWriter();
>>>
>>> return position;
>>> }
>>>
>>> @Override
>>> public long getPos() {
>>> Preconditions.checkNotNull(writer);
>>> return position + writer.getDataSize();
>>> }
>>>
>>> @Override
>>> public void close() throws IOException {
>>> if (writer != null) {
>>> writer.close();
>>> writer = null;
>>> }
>>> }
>>>
>>> @Override
>>> public void write(T element) throws IOException {
>>> Preconditions.checkNotNull(writer);
>>> writer.write(element);
>>> }
>>>
>>> @Override
>>> public Writer<T> duplicate() {
>>> return new FlinkProtoParquetWriter<>(protoClass);
>>> }
>>>
>>> private ParquetWriter<T> createWriter() throws IOException {
>>> return ProtoParquetWriter
>>> .<T>builder(path)
>>> .withPageSize(pageSize)
>>> .withCompressionCodec(compressionCodecName)
>>> .withProtoClass(protoClass)
>>> .build();
>>> }
>>> }
>>>
>>>
>>> Rafi
>>>
>>>
>>>
