We have an ELT use case where we ingest data from Kafka tran Apache Flink #troubleshooting

We have an ELT use-case where we ingest data from ...

Sameer Chandra

07/12/2023, 8:08 AM

We have an ELT use-case where we ingest data from Kafka, transform it, and write to S3 in parquet format. Since we use a CheckpointRollingPolicy, files are written to s3 on checkpoint (15 min interval). We use rocksdb as state backend. Until the checkpoint is triggered, are these records stored in JVM heap?

Rashmin Patel

07/12/2023, 9:37 AM

As per my knowledge, records are not buffered locally when using streaming file sink. They are continuously written to s3 in form of part files. Only thing which happens at checkpoint trigger is, these part files are marked committed by moving them from in-progress state to completed. Reference: https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/datastream/streamfile_sink/

Sameer Chandra

07/12/2023, 9:39 AM

@Rashmin Patel We are using Bulk Encoded format rather than row encoded. In this case, there shouldn't be any in-progress files present as per my understanding

👍 1

Rashmin Patel

07/12/2023, 9:48 AM

This should depend on the how frequently flush is being called by BulkWriter. Instead of flushing record by record, BulkWriter would make a batch of bunch of records, encode them and then flush a block.

Sameer Chandra

07/12/2023, 9:55 AM

Exactly. BulkWriter can only have oncheckpointrollingpolicy right? Which means the records will get flushed at each checkpoint if I am not wrong

Rashmin Patel

07/12/2023, 9:59 AM

Having OnCheckpointRollingPolicy does not restrict the BulkWriter from doing intermediate flushing.

Rashmin Patel

07/12/2023, 10:03 AM

Let's say a writer is flushing in batch size of 10 records, and sink receives total of 247 records in interval of 15 mins, then total number of flushes would be 25. (24 flushes with 10 records each and last flush with 7 records which got pre-triggered due to checkpoint trigger)

Rashmin Patel

07/12/2023, 10:04 AM

As per official docs

Copy code

The BulkWriter logic defines how new elements are added and flushed, and how a batch of records is finalized for further encoding purposes.

Rashmin Patel

07/12/2023, 10:06 AM

And data buffered across flushes (which is 10 records in this example) will reside on JVM heap, as per my understanding.

Sameer Chandra

07/12/2023, 10:16 AM

The question is, where do they flush it to? No temp files are created between checkpoints

Rashmin Patel

07/12/2023, 11:10 AM

For Parquet format, as per code, I can see that flush is happening to disk. You can check

InternalParquetRecordWriter

class

Rashmin Patel

07/12/2023, 12:00 PM

@Sameer Chandra Here disk refers to our storage destination which is S3 in this case. On diving a bit deeper, I found the reason of why it doesn't create any in-progress files in case of S3. It uses

S3Committer

class internally for doing multi-part upload, and commit happens in alignment with checkpoint, hence we don't see any in-progress file, but the final object on commit. If you carefully see the "*Last modified*" timestamp of any part file, then it would be

checkpoint interval

behind the end of window, which means S3 started receiving data as soon as the window started. I hope this explanation answers your question. Note: "*Last modified*" in case of multi-part upload is the upload start timestamp. (Ref)

👍 3

Sameer Chandra

07/12/2023, 2:56 PM

Makes sense! Thanks Rashmin!

Open in Slack

Previous Next